{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "本章将介绍如何用muduo网络库完成常见的TCP网络编程任务。\n",
    "内容如下：\n",
    "1. [UNP]中的五个简单协议，包括echo、daytime、time、discard、chargen等。\n",
    "2. 文件传输，示范非阻塞TCP网络程序中如何完整地发送数据。\n",
    "3. Boost.Asio中的示例，包括timer2～6、chat等。chat实现了TCP封包与拆包(codec)。\n",
    "4. muduo Buffer class的设计与使用。\n",
    "5. Protobuf编码解码器(codec)与消息分发器(dispatcher)。\n",
    "6. 限制服务器的最大并发连接数。\n",
    "7. Java Netty中的示例，包括discard、echo、uptime等，其中的discard和echo带流量统计功能。\n",
    "8. 用于测试两台机器的往返延迟的roundtrip。\n",
    "9. 用timing wheel踢掉空闲连接。\n",
    "10. 一个基于TCP的应用层广播hub。\n",
    "11. 云风的串并转换连接服务器multiplexer，及其自动化测试。\n",
    "12. socks4a代理服务器，包括简单的TCP中继(relay)。\n",
    "13. 一个提供短址服务的httpd服务器。\n",
    "14. 与其他库的集成，包括UDNS、c-ares DNS、curl等等。\n",
    "\n",
    "这些例子都比较简单，逻辑不复杂，代码也很短，适合摘取关键部分放到博客上。其中一些有一定的代表性与针对性，比如“如何传输完整的文件”估计是网络编程的初学者经常遇到的问题。请注意，muduo是设计来开发内网的网络程序，它没有做任何安全方面的加强措施，如果用在公网上可能会受到攻击，在后面的例子中我会谈到这一点。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 7.1 五个简单TCP示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "本节将介绍五个简单TCP网络服务程序，包括echo(RFC 862)、discard(RFC 863)、chargen(RFC 864)、daytime(RFC 867)、time(RFC 868)这五个协议，以及time协议的客户端。\n",
    "\n",
    "各程序的协议简介如下。\n",
    "- discard：丢弃所有收到的数据。\n",
    "- daytime：服务端accept连接之后，以字符串形式发送当前时间，然后主动断开连接。\n",
    "- time：服务端accept连接之后，以二进制形式发送当前时间（从Epoch到现在的秒数）​，然后主动断开连接；我们需要一个客户程序来把收到的时间转换为字符串。\n",
    "- echo：回显服务，把收到的数据发回客户端。\n",
    "- chargen：服务端accept连接之后，不停地发送测试数据。\n",
    "\n",
    "以上五个协议使用不同的端口，可以放到同一个进程中实现，且不必使用多线程。完整的代码见[muduo/examples/simple](../muduo/examples/)。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## discard\n",
    "\n",
    "discard恐怕算是最简单的长连接TCP应用层协议，它只需要关注“三个半事件”中的“消息/数据到达”事件，事件处理函数如下：\n",
    "\n",
    "- [discard](./discard/discard.h)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## daytime\n",
    "\n",
    "daytime是短连接协议，在发送完当前时间后，由服务端主动断开连接。它只需要关注“三个半事件”中的“连接已建立”事件，事件处理函数如下：\n",
    "\n",
    "- [daytime](./daytime/daytime.h)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! daytime\n",
    "20251011 22:33:10.715062Z 63747 INFO  pid = 63747 - main.cc:13\n",
    "20251011 22:33:42.451771Z 63747 INFO  TcpServer::newConnection [DaytimeServer] - new connection [DaytimeServer-0.0.0.0:2013#1] from 127.0.0.1:40094 - TcpServer.cc:80\n",
    "20251011 22:33:42.451861Z 63747 INFO  DaytimeServer - 127.0.0.1:40094 -> 127.0.0.1:2013 is UP - daytime.h:47\n",
    "20251011 22:53:52.250178Z 63747 INFO  DaytimeServer - 127.0.0.1:40094 -> 127.0.0.1:2013 is DOWN - daytime.h:47\n",
    "20251011 22:53:52.250221Z 63747 INFO  TcpServer::removeConnectionInLoop [DaytimeServer] - connection DaytimeServer-0.0.0.0:2013#1 - TcpServer.cc:109"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! nc 127.0.0.1 2013\n",
    "20251011 22:33:42.451890"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## time\n",
    "\n",
    "time协议与daytime极为类似，只不过它返回的不是日期时间字符串，而是一个32-bit整数，表示从`1970-01-0100：00：00Z`到现在的秒数。当然，这个协议有“2038年问题”​。服务端只需要关注“三个半事件”中的“连接已建立”事件，事件处理函数如下\n",
    "\n",
    "- [time.h](./time/timeserver.h)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "00000000  68 ea eb 0b                                       |h...|\n",
      "00000004\n"
     ]
    }
   ],
   "source": [
    "! nc -v -w 1 127.0.0.1 2037 | hexdump -C"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! time\n",
    "20251011 23:35:55.574694Z 78035 INFO  pid = 78035 - main.cc:13\n",
    "20251011 23:35:59.492157Z 78035 INFO  TcpServer::newConnection [TimeServer] - new connection [TimeServer-0.0.0.0:2037#1] from 127.0.0.1:49268 - TcpServer.cc:80\n",
    "20251011 23:35:59.492240Z 78035 INFO  TimeServer - 127.0.0.1:49268 -> 127.0.0.1:2037 is UP - timeserver.h:46\n",
    "20251011 23:37:53.269305Z 78035 INFO  TimeServer - 127.0.0.1:49268 -> 127.0.0.1:2037 is DOWN - timeserver.h:46\n",
    "20251011 23:37:53.269369Z 78035 INFO  TcpServer::removeConnectionInLoop [TimeServer] - connection TimeServer-0.0.0.0:2037#1 - TcpServer.cc:109"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## time客户端\n",
    "\n",
    "因为time服务端发送的是二进制数据，不便直接阅读，我们编写一个客户端来解析并打印收到的4个字节数据。这个程序只需要关注“三个半事件”中的“消息/数据到达”事件，事件处理函数如下：\n",
    "\n",
    "- [timeclient.cc](./timeclient/timeclient.cc)\n",
    "\n",
    "注意其中考虑到了如果数据没有一次性收全，已经收到的数据会累积在`Buffer`里（在`else`分支里没有调用`Buffer::retrieve`＊系列函数）​，以等待后续数据到达，程序也不会阻塞。这样即便服务器一个字节一个字节地发送数据，代码还是能正常工作，这也是非阻塞网络编程必须在用户态使用接收缓冲的主要原因。\n",
    "\n",
    "这是我们第一次用到`TcpClient class`\n",
    "\n",
    "- [timeclient.cc](./timeclient/timeclient.cc)\n",
    "\n",
    "注意TcpConnection对象表示“一次”TCP连接，连接断开之后不能重建。TcpClient重试之后新建的连接会是另一个TcpConnection对象。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! ./time\n",
    "20251012 00:20:56.833630Z 83261 INFO  pid = 83261 - main.cc:13\n",
    "20251012 00:23:15.651988Z 83261 INFO  TcpServer::newConnection [TimeServer] - new connection [TimeServer-0.0.0.0:2037#1] from 127.0.0.1:46186 - TcpServer.cc:80\n",
    "20251012 00:23:15.652104Z 83261 INFO  TimeServer - 127.0.0.1:46186 -> 127.0.0.1:2037 is UP - timeserver.h:46\n",
    "20251012 00:23:15.652407Z 83261 INFO  TimeServer - 127.0.0.1:46186 -> 127.0.0.1:2037 is DOWN - timeserver.h:46\n",
    "20251012 00:23:15.652449Z 83261 INFO  TcpServer::removeConnectionInLoop [TimeServer] - connection TimeServer-0.0.0.0:2037#1 - TcpServer.cc:109"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "20251012 00:23:15.651079Z 83718 INFO  pid = 83718 - timeclient.cc:70\n",
      "20251012 00:23:15.651495Z 83718 INFO  TcpClient::TcpClient[TimeClient] - connector 0x5F197B9E82F0 - TcpClient.cc:69\n",
      "20251012 00:23:15.651501Z 83718 INFO  TcpClient::connect[TimeClient] - connecting to 127.0.0.1:2037 - TcpClient.cc:107\n",
      "20251012 00:23:15.651933Z 83718 INFO  127.0.0.1:46186 -> 127.0.0.1:2037 is UP - timeclient.cc:39\n",
      "20251012 00:23:15.652237Z 83718 INFO  Server time = 1760228595, 20251012 00:23:15.000000 - timeclient.cc:58\n",
      "20251012 00:23:15.652284Z 83718 INFO  127.0.0.1:46186 -> 127.0.0.1:2037 is DOWN - timeclient.cc:39\n",
      "20251012 00:23:15.652384Z 83718 INFO  TcpClient::~TcpClient[TimeClient] - connector 0x5F197B9E82F0 - TcpClient.cc:75\n"
     ]
    }
   ],
   "source": [
    "! cd ../run && ./timeclient 127.0.0.1 "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## echo\n",
    "\n",
    "前面几个协议都是单向接收或发送数据，echo是我们遇到的第一个双向的协议：服务端把客户端发过来的数据原封不动地传回去。它只需要关注“三个半事件”中的“消息/数据到达”事件，事件处理函数已在p.139列出，这里复制一遍。\n",
    "\n",
    "- [echo.cc](../CH06/echo/echo.cc)\n",
    "\n",
    "- 练习1：修改`EchoServer::onMessage()`，实现大小写互换。\n",
    "- 练习2：修改`EchoServer::onMessage()`，实现[ROT13](https://en.wikipedia.org/wiki/ROT13)加密"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## charge\n",
    "\n",
    "Chargen协议很特殊，它只发送数据，不接收数据。而且，它发送数据的速度不能快过客户端接收的速度，因此需要关注“三个半事件”中的半个“消息/数据发送完毕”事件(`onWriteComplete`)，事件处理函数如下：\n",
    "\n",
    "- [chargen.h](./chargen/chargen.h)\n",
    "\n",
    "完整的chargen服务端还带流量统计功能，用到了定时器，我们会在§ 7.8介绍定时器的使用，到时候再回头来看相关代码"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! nc localhost 2019 | head\n",
    "!\"#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefgh\n",
    "\"#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghi\n",
    "#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghij\n",
    "$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijk\n",
    "%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijkl\n",
    "&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklm\n",
    "'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmn\n",
    "()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmno\n",
    ")*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnop\n",
    "*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopq"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! ./chargen \n",
    "20251012 00:51:15.976325Z 88022 INFO  pid = 88022 - main.cc:13\n",
    "0.000 MiB/s\n",
    "0.000 MiB/s\n",
    "0.000 MiB/s\n",
    "20251012 00:52:43.346505Z 88022 INFO  TcpServer::newConnection [ChargenServer] - new connection [ChargenServer-0.0.0.0:2019#1] from 127.0.0.1:56196 - TcpServer.cc:80\n",
    "20251012 00:52:43.346565Z 88022 INFO  ChargenServer - 127.0.0.1:56196 -> 127.0.0.1:2019 is UP - chargen.h:75\n",
    "0.051 MiB/s\n",
    "0.000 MiB/s\n",
    "0.000 MiB/s\n",
    "0.000 MiB/s\n",
    "0.000 MiB/s"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 五合一\n",
    "\n",
    "前面五个程序都用到了EventLoop。这其实是个Reactor，用于注册和分发IO事件。*muduo遵循one loop per thread模型，多个服务端(TcpServer)和客户端(TcpClient)可以共享同一个EventLoop，也可以分配到多个EventLoop上以发挥多核多线程的好处*。\n",
    "\n",
    "这里我们把五个服务端用同一个EventLoop跑起来，程序还是单线程的，功能却强大了很多：\n",
    "\n",
    "- [allinone.cc](./allinone/allinone.cc)\n",
    "\n",
    "![7.1a](./images/7.1a.png)\n",
    "\n",
    "这个例子充分展示了Reactor模式复用线程的能力，让一个单线程程序同时具备多个网络服务功能。一个容易想到的例子是httpd同时侦听80端口和443端口，另一个例子是程序中有多个TcpClient，分别和数据库、Redis、Sudoku Solver等后台服务打交道。对于初次接触这种编程模型的读者，值得跟踪代码运行的详细过程，弄清楚每个事件每个回调发生的时机与条件。\n",
    "\n",
    "以上几个协议的消息格式都非常简单，没有涉及TCP网络编程中常见的分包处理，在后文§ 7.3讲Boost.Asio的聊天服务器时我们再来讨论这个问题。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 7.2 文件传输"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "本节用发送文件的例子来说明`TcpConnection::send()`的使用。到目前为止，我们用到了`TcpConnection::send()`的两个重载，分别是`send(const string&)`和`send(const void* message，size_t len)`。\n",
    "\n",
    "- [TcpConnection.h](../muduo/muduo/net/TcpConnection.h)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在非阻塞网络编程中，发送消息通常是由网络库完成的，用户代码不会直接调用`write(2)`或`send(2)`等系统调用。原因见p.205-7.4.2“TcpConnection必须要有output buffer”​。\n",
    "\n",
    "在使用`TcpConnection::send()`时值得注意的有几点：\n",
    "\n",
    "- `send()`的返回类型是`void`，意味着用户不必关心调用`send()`时成功发送了多少字节，muduo库会保证把数据发送给对方。\n",
    "- `send()`是线程安全、原子的。多个线程可以同时调用`send()`，消息之间不会混叠或交织。但是多个线程同时发送的消息(s)的先后顺序是不确定的，muduo只能保证每个消息本身的完整性。另外，`send()`在多线程下仍然是非阻塞的。\n",
    "> 假设两个线程同时各自发送了一条任意长度的消息，那么这两条消息a、b的发送顺序要么是先a后b，要么是先b后a，不会出现\"a的前一半，b，a的后一半\"这种交织情况。\n",
    "- `send(const void* message，size_t len)`这个重载最平淡无奇，可以发送任意字节序列。\n",
    "- `send(const StringPiece& message)`这个重载可以发送`std::string`和`const char*`，其中[`StringPiece`](../muduo/muduo/base/StringPiece.h) 是Google发明的专门用于传递字符串参数的class，这样程序里就不必为`const char*`和`const std::string&`提供两份重载了。\n",
    "- `send(Buffer*)`有点特殊，它以指针为参数，而不是常见的`const`引用，因为函数中可能用`Buffer::swap()`来高效地交换数据，避免内存拷贝，起到类似C++右值引用的效果。\n",
    "> 目前的实现尚未照此办理\n",
    "- 如果将来支持C++11，那么可以增加对右值引用的重载，这样可以用move语义来避免内存拷贝。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "下面我们来实现一个发送文件的命令行小工具，这个工具的协议很简单，在启动时通过命令行参数指定要发送的文件，然后在2021端口侦听，每当有新连接进来，就把文件内容完整地发送给对方。\n",
    "\n",
    "如果不考虑并发，那么这个功能用netcat加重定向就能实现。这里展示的版本更加健壮，比方说发送100MB的文件，支持上万个并发客户连接；内存消耗只与并发连接数有关，跟文件大小无关；任何连接可以在任何时候断开，程序不会有内存泄漏或崩溃。\n",
    "> 我用Java实现了压力测试，代码位于[examples/filetransfer/loadtest](../muduo/examples/filetransfer/loadtest/)\n",
    "\n",
    "我一共写了三个版本，代码位于[examples/filetransfer](../muduo/examples/filetransfer)\n",
    "\n",
    "1. **一次性发送版本**\n",
    "   - 实现方式：一次性把文件读入内存，调用 `send(const string&)` 发送完毕\n",
    "   - 健壮性评估：满足除“内存消耗只与并发连接数有关，跟文件大小无关”之外的所有健壮性要求\n",
    "\n",
    "2. **分块发送版本**\n",
    "   - 实现方式：一块一块地发送文件，减少内存使用\n",
    "   - 关键技术：使用 `WriteCompleteCallback`\n",
    "   - 健壮性评估：满足上述全部健壮性要求\n",
    "\n",
    "3. **智能指针管理版本**\n",
    "   - 实现方式：同版本2，但采用 `shared_ptr` 管理 `FILE*`\n",
    "   - 技术优势：避免手动调用 `::fclose(3)`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 一次性发送版本\n",
    "\n",
    "在建立好连接之后，把文件的全部内容读入一个`string`，一次性调用`TcpConnection::send()`发送。不用担心文件发送不完整。也不用担心`send()`之后立刻`shutdown()`会有什么问题，见下一节的说明。\n",
    "\n",
    "- [download.cc](./download/download.cc)\n",
    "\n",
    "注意每次建立连接的时候我们都去重新读一遍文件，这是考虑到文件有可能被其他程序修改。如果文件是`immutable`的，整个程序就可以共享同一个`fileContent`对象。\n",
    "\n",
    "这个版本有一个明显的缺陷，即内存消耗与（并发连接数×文件大小）成正比，文件越大内存消耗越多，如果文件大小上GB，那几乎就是灾难了。只需要建立少量并发连接就能把服务器的内存耗尽，因此我们有了版本二。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 分块发送版本\n",
    "\n",
    "为了解决版本一占用内存过多的问题，我们采用流水线的思路，当新建连接时，先发送文件的前64KiB数据，等这块数据发送完毕时再继续发送下64KiB数据，如此往复直到文件内容全部发送完毕。\n",
    "\n",
    "代码中使用了`TcpConnection::setContext()`和`getContext()`来保存`TcpConnection`的用户上下文（这里是`FILE*`）​，因此不必使用额外的`std::map<TcpConnectionPtr，FILE*>`来记住每个连接的当前文件位置。\n",
    "\n",
    "- [download2.cc](./download2/download2.cc)\n",
    "\n",
    "注意每次建立连接的时候我们都去重新打开那个文件，使得程序中文件描述符的数量翻倍（每个连接占一个`socket fd`和一个`file fd`）​，这是考虑到文件有可能被其他程序修改。如果文件是`immutable`的，一种改进措施是：整个程序可以共享同一个文件描述符，然后每个连接记住自己当前的偏移量，在`onWriteComplete()`回调函数里用`pread(2)`来读取数据\n",
    "\n",
    "> `pread(2)`: 用于从文件描述符的指定位置读取数据，不会改变文件的偏移量\n",
    "\n",
    "这个版本也存在一个问题，如果客户端故意只发起连接，不接收数据，那么要么把服务器进程的文件描述符耗尽，要么占用很多服务端内存（因为每个连接有64KiB的发送缓冲区）​。解决办法可参考后文§ 7.7“限制服务器的最大并发连接数”和§ 7.10“用timing wheel踢掉空闲连接”​。必须说明的是，muduo并不是设计来编写面向公网的网络服务程序，这种服务程序需要在安全性方面下很多工夫，我个人对此不在行，我更关心实现内网（不一定是局域网）的高效服务程序。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```txt\n",
    "1. onConnection() \n",
    "   └── 发送第1块数据\n",
    "        │\n",
    "2. 第1块数据完全写入TCP缓冲区\n",
    "   └── 触发 onWriteComplete()\n",
    "        │\n",
    "3. onWriteComplete()\n",
    "   └── 发送第2块数据\n",
    "        │\n",
    "4. 第2块数据完全写入TCP缓冲区\n",
    "   └── 触发 onWriteComplete()\n",
    "        │\n",
    "5. 重复直到文件结束\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 智能指针管理版本\n",
    "\n",
    "用`shared_ptr`的custom deleter来减轻资源管理负担，使得`FILE*`的生命期和TcpConnection一样长，代码也更简单了。\n",
    "\n",
    "- [download3.cc](./download3/download3.cc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! diff -u download2/download2.cc download3/download3.cc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*体现了现代C++的资源管理思路，即无须手动释放资源，而是通过将资源与对象生命期绑定，在对象析构的时候自动释放资源，从而把资源管理转换为对象生命期管理，而后者是早已解决了的问题。这正是C++最重要的编程技法：RAII。*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## *为什么`TcpConnection::shutdown()`没有直接关闭TCP连接?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "我曾经收到一位网友的来信：​“在simple的[daytime](./daytime/daytime.h)示例中，服务端主动关闭时调用的是如下函数序列，这不是只是关闭了连接上的写操作吗，怎么是关闭了整个连接？​”\n",
    "\n",
    "**笔者答复如下**：\n",
    "\n",
    "`muduo TcpConnection`没有提供`close()`，而只提供`shutdown()`，这么做是为了收发数据的完整性。\n",
    "\n",
    "TCP是一个全双工协议，同一个文件描述符既可读又可写，`shutdownWrite()`关闭了“写”方向的连接，保留了“读”方向，这称为TCP half-close。如果直接`close(socket_fd)`，那么`socket_fd`就不能读或写了。\n",
    "\n",
    "用`shutdown`而不用`close`的效果是，如果对方已经发送了数据，这些数据还“在路上”​，那么muduo不会漏收这些数据。换句话说，muduo在TCP这一层面解决了“当你打算关闭网络连接的时候，如何得知对方是否发了一些数据而你还没有收到？​”这一问题。当然，这个问题也可以在上面的协议层解决，双方商量好不再互发数据，就可以直接断开连接。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> - **`close(fd)`**: 直接把这个文件描述符的读写能力都关闭了；在 TCP 上等价于同时停止发送与接收。调用后应用层就不能再 `read` 或 `write`。如果此时对端仍有数据在网络上飞行，本端在内核缓冲区里还没收到，就会被丢掉。\n",
    "> - **`shutdown(fd, SHUT_WR)`**: 只关闭写方向（发送方向），保留读能力。这个动作会触发 TCP 协议栈发一个 FIN 报文告诉对端“我这边已经发完了，不会再写数据给你；但我还能继续读你发来的数据”。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "练习：阅读代码，回答“如果被动关闭连接，muduo的行为如何？​”"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "另外，如果当前`output buffer`里还有数据尚未发出的话，muduo也不会立刻调用`shutdownWrite`，而是等到数据发送完毕再`shutdown`，可以避免对方漏收数据。\n",
    "\n",
    "- [TcpConnection.cc](../muduo/muduo/net/TcpConnection.cc#L375)\n",
    "\n",
    "muduo这种关闭连接的方式对对方也有要求，那就是对方`read()`到`0`字节之后会主动关闭连接（无论`shutdownWrite()`还是`close()`）​，一般的网络程序都会这样，不是什么问题。当然，这么做有一个潜在的安全漏洞，万一对方故意不关闭连接，那么muduo的连接就一直半开着，消耗系统资源。必要时可以调用`TcpConnection::handleClose()`来强行关闭连接，这需要将`handleClose()`改为`public`成员函数。\n",
    "\n",
    "完整的流程见图7-1。我们发完了数据，于是`shutdownWrite`，发送`TCP FIN`分节，对方会读到`0`字节，然后对方通常会关闭连接。这样muduo会读到`0`字节，然后muduo关闭连接。​（思考题：在`shutdown()`之后，muduo回调`connection callback`的时间间隔大约是一个round-trip time，为什么？​）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"./images/7.1.png\" alt=\"7.1\" style=\"zoom:35%;\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "如果有必要，对方可以在`read()`返回`0`之后继续发送数据，这是直接利用了half-close TCP连接。muduo不会漏收这些数据。\n",
    "\n",
    "那么muduo什么时候真正close socket呢？在`TcpConnection`对象析构的时候。`TcpConnection`持有一个Socket对象，Socket是一个RAII handler，它的析构函数会 。这样，如果发生`TcpConnection`对象泄漏，那么我们从`/proc/pid/fd/`就能找到没有关闭的文件描述符，便于查错。\n",
    "\n",
    "muduo在`read()`返回`0`的时候会回调`connection callback`，TcpServer或Tcp-Client把`TcpConnection`的引用计数减一。如果引用计数降到零，则表明用户代码也不持有`TcpConnection`，它就会析构了。\n",
    "\n",
    "参考：​《TCP/IP详解》​[TCPv1]18.5节“TCP Half-Close”和《UNIX网络编程（第3版）​》​[UNP]6.6节“`shutdown()`函数”​。\n",
    "\n",
    "在网络编程中，应用程序发送数据往往比接收数据简单（实现非阻塞网络库正相反，发送比接收难）​，下一节我们再谈接收并解析消息的要领。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "区分 应用代码 和 网络库内部实现 两个视角下“收发哪个更难”的差异\n",
    "\n",
    "| 视角                 | 发送（Send）                                                 | 接收（Receive）                                              | 难度对比结论                             |\n",
    "| -------------------- | ------------------------------------------------------------ | ------------------------------------------------------------ | ---------------------------------------- |\n",
    "| **应用程序**         | - 直接 `send()/write()` 把数据交给内核即可。<br>- 协议格式由自己构造，可一次性完成。 | - 需要把字节流拆分成协议消息。<br>- 要处理粘包、半包、超时、校验等逻辑。 | 接收更难：解析与状态管理复杂。           |\n",
    "| **非阻塞网络库实现** | - 需处理部分写、写缓冲满 (`EAGAIN`)。<br>- 维护输出缓冲与高水位限流，避免背压失控。<br>- 保证线程安全，把应用数据排入事件循环。<br>- 在关闭前确保缓冲数据全部发出。 | - 主要处理读事件触发、`read()` 结果为 0 或 `EAGAIN`。<br>- 将字节投递给应用回调即可，逻辑相对线性。 | 发送更难：需要复杂的缓冲管理与事件调度。 |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# * 7.3 Boost.Asio的聊天服务器"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "本节将介绍一个与Boost.Asio的示例代码中的聊天服务器功能类似的网络服务程序，包括客户端与服务端的muduo实现。这个例子的主要目的是介绍如何处理分包，并初步涉及muduo的多线程功能。本文的代码位于[examples/asio/chat/](../muduo/examples/asio/chat/)。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## TCP分包\n",
    "\n",
    "§ 7.1“五个简单TCP示例”中处理的协议没有涉及分包，在TCP这种字节流协议上做应用层分包是网络编程的基本需求。分包指的是在发生一个消息(message)或一帧(frame)数据时，通过一定的处理，让接收方能从字节流中识别并截取（还原）出一个个消息。​“粘包问题”是个伪问题。\n",
    "\n",
    "对于短连接的TCP服务，分包不是一个问题，只要发送方主动关闭连接，就表示一条消息发送完毕，接收方`read()`返回`0`，从而知道消息的结尾。例如§ 7.1里的daytime和time协议。\n",
    "\n",
    "对于长连接的TCP服务，分包有四种方法：\n",
    "1. 消息长度固定，比如muduo的roundtrip示例就采用了固定的16字节消息。\n",
    "2. 使用特殊的字符或字符串作为消息的边界，例如HTTP协议的headers以“\\r\\n”为字段的分隔符。\n",
    "3. 在每条消息的头部加一个长度字段，这恐怕是最常见的做法，本文的聊天协议也采用这一办法。\n",
    "4. 利用消息本身的格式来分包，例如XML格式的消息中`<root>...</root>`的配对，或者JSON格式中的`{...}`的配对。解析这种消息格式通常会用到状态机(state machine)\n",
    "\n",
    "在后文的代码讲解中还会仔细讨论用长度字段分包的常见陷阱。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 聊天服务\n",
    "\n",
    "本节实现的聊天服务非常简单，由服务端程序和客户端程序组成，协议如下：\n",
    "\n",
    "- 服务端程序在某个端口侦听 (listen) 新的连接。\n",
    "- 客户端向服务端发起连接。\n",
    "- 连接建立之后，客户端随时准备接收服务端的消息并在屏幕上显示出来。\n",
    "- 客户端接受键盘输入，以回车为界，把消息发送给服务端。\n",
    "- 服务端接收到消息之后，依次发送给每个连接到它的客户端；原来发送消息的客户端进程也会收到这条消息。\n",
    "- 一个服务端进程可以同时服务多个客户端进程。当有消息到达服务端后，每个客户端进程都会收到同一条消息，服务端广播发送消息的顺序是任意的，不一定哪个客户端会先收到这条消息。\n",
    "- （可选）如果消息 A 先于消息 B 到达服务端，那么每个客户端都会先收到 A 再收到 B。\n",
    "\n",
    "这实际上是一个简单的基于TCP的应用层广播协议，由服务端负责把消息发送给每个连接到它的客户端。参与“聊天”的既可以是人，也可以是程序。在后文§7.11中，我将介绍一个稍微复杂一点的例子hub，它有“聊天室”的功能，客户端可以注册特定的topic(s)，并往某个topic发送消息，这样代码更有意思。\n",
    "\n",
    "我在“谈一谈网络编程学习经验”​（附录A）中把聊天服务列为“最主要的三个例子”之一，其与前面的“五个简单TCP协议”不同，聊天服务的特点是“连接之间的数据有交流，从a连接收到的数据要发给b连接。这样对连接管理提出了更高的要求：如何用一个程序同时处理多个连接？`fork()-per-connection`似乎是不行的。如何防止串话？b有可能随时断开连接，而新建立的连接c可能恰好复用了b的文件描述符，那么a会不会错误地把消息发给c？​”muduo的这个例子充分展示了解决以上问题的手法。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 消息格式\n",
    "\n",
    "本聊天服务的消息格式非常简单，​“消息”本身是一个字符串，每条消息有一个4字节的头部，以网络序存放字符串的长度。消息之间没有间隙，字符串也不要求以'\\0'结尾。比方说有两条消息“hello”和“chenshuo”​，那么打包后的字节流共有21字节：\n",
    "\n",
    "```txt\n",
    "0x00, 0x00, 0x00, 0x05, 'h', 'e', 'l', 'l', 'o',\n",
    "0x00, 0x00, 0x00, 0x08, 'c', 'h', 'e', 'n', 's', 'h', 'u', 'o'\n",
    "```\n",
    "\n",
    "打包的代码 这段代码把`string message`打包为`muduo::net::Buffer`，并通过`conn`发送。由于这个codec的代码位于头文件中，因此反复出现了`muduo::net`namespace。\n",
    "\n",
    "- [codec.h](./codec/codec.h#L55)\n",
    "\n",
    "以前面提到的两条消息的字节流为例：\n",
    "1. 每次收到一个字节的数据，`onMessage()`被调用21次；\n",
    "2. 数据分两次到达，第一次收到2个字节，不足消息的长度字段；\n",
    "3. 数据分两次到达，第一次收到4个字节，刚好够长度字段，但是没有body；\n",
    "4. 数据分两次到达，第一次收到8个字节，长度完整，但body不完整；\n",
    "5. 数据分两次到达，第一次收到9个字节，长度完整，body也完整；\n",
    "6. 数据分两次到达，第一次收到10个字节，第一条消息的长度完整、body也完整，第二条消息长度不完整；\n",
    "7. 请自行移动和增加分割点，验证各种情况；一共有超过100万种可能($2^{21-1}$)。\n",
    "8. 数据一次就全部到达，这时必须用`while`循环来读出两条消息，否则消息会堆积在`Buffer`中。\n",
    "\n",
    "请读者验证`onMessage()`是否做到了以上几点。这个例子充分说明了non-blocking read必须和input buffer一起使用。而且在写decoder的时候一定要在收到完整的消息(L40)之后再retrieve整条消息（L42和L45）​，除非接收方使用复杂的状态机来解码。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 编解码器LengthHeaderCodec\n",
    "\n",
    "[有人评论](https://www.cnblogs.com/Solstice/archive/2011/02/02/1948839.html#2022206)muduo的接收缓冲区不能设置回调函数的触发条件，确实如此。每当socket可读时，muduo的`TcpConnection`会读取数据并存入input buffer，然后回调用户的函数。不过，一个*简单的间接层*就能解决问题，让用户代码只关心“消息到达”而不是“数据到达”​，如本例中的`LengthHeaderCodec`所展示的那样。\n",
    "\n",
    "- [codec.h](./codec/codec.h#L64)\n",
    "\n",
    "这段代码把以`Buffer*`为参数的`MessageCallback`转换成了以`const string&`为参数的`StringMessageCallback`，让用户代码不必关心分包操作，具体的调用时序图见p.231图7-29。如果编程语言相同，客户端和服务端可以（应该）共享同一个codec，这样既节省工作量，又避免因对协议理解不一致而导致的错误。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 服务端的实现\n",
    "\n",
    "聊天服务器的服务端代码小于100行，不到asio的一半\n",
    "\n",
    "- [server.cc](./codec/server.cc#L64)\n",
    "\n",
    "如果你读过asio的对应代码，会不会觉得Reactor往往比Proactor容易使用？"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## * 客户端的实现\n",
    "\n",
    "我有时觉得服务端的程序常常比客户端的更容易写，聊天服务器再次验证了我的看法。客户端的复杂性来自于它要读取键盘输入，而`EventLoop`是独占线程的，所以我用了两个线程：`main()`函数所在的线程负责读键盘，另外用一个`EventLoopThread`来处理网络IO。\n",
    "> 我暂时没有把标准输入输出融入Reactor的想法，因为服务器程序很少用到`stdin`和`stdout`。\n",
    "\n",
    "现在来看代码\n",
    "\n",
    "- [client.c](./codec/client.cc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 简单测试"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "这样就有两个客户端进程参与聊天。在第二个窗口里输入一些字符并回车，字符会出现在本窗口和第三个窗口中。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "$ ./asio_chat_server 3000\n",
    "20251013 01:25:30.462113Z  6300 INFO  pid = 6300 - server.cc:85\n",
    "20251013 01:25:40.108960Z  6300 INFO  TcpServer::newConnection [ChatServer] - new connection [ChatServer-0.0.0.0:3000#1] from 127.0.0.1:47468 - TcpServer.cc:80\n",
    "20251013 01:25:40.109055Z  6300 INFO  127.0.0.1:47468 -> 127.0.0.1:3000 is UP - server.cc:49\n",
    "20251013 01:25:47.594157Z  6300 INFO  TcpServer::newConnection [ChatServer] - new connection [ChatServer-0.0.0.0:3000#2] from 127.0.0.1:52198 - TcpServer.cc:80\n",
    "20251013 01:25:47.594222Z  6300 INFO  127.0.0.1:52198 -> 127.0.0.1:3000 is UP - server.cc:49"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "$ ./asio_chat_client 127.0.0.1 3000\n",
    "20251013 01:25:40.107545Z  6378 INFO  pid = 6378 - client.cc:86\n",
    "20251013 01:25:40.108678Z  6378 INFO  TcpClient::TcpClient[ChatClient] - connector 0x59EC328354A0 - TcpClient.cc:69\n",
    "20251013 01:25:40.108700Z  6378 INFO  TcpClient::connect[ChatClient] - connecting to 127.0.0.1:3000 - TcpClient.cc:107\n",
    "20251013 01:25:40.108978Z  6379 INFO  127.0.0.1:47468 -> 127.0.0.1:3000 is UP - client.cc:54\n",
    "<<< hello\n",
    "A\n",
    "<<< A\n",
    "<<< B"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "$ ./asio_chat_client 127.0.0.1 3000\n",
    "20251013 01:25:47.593628Z  6419 INFO  pid = 6419 - client.cc:86\n",
    "20251013 01:25:47.593938Z  6419 INFO  TcpClient::TcpClient[ChatClient] - connector 0x609BFC7D84A0 - TcpClient.cc:69\n",
    "20251013 01:25:47.593960Z  6419 INFO  TcpClient::connect[ChatClient] - connecting to 127.0.0.1:3000 - TcpClient.cc:107\n",
    "20251013 01:25:47.594163Z  6420 INFO  127.0.0.1:52198 -> 127.0.0.1:3000 is UP - client.cc:54\n",
    "hello\n",
    "<<< hello\n",
    "<<< A\n",
    "B\n",
    "<<< B\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "代码示例中还有另外三个server程序，都是多线程的，详细介绍在p.260。\n",
    "\n",
    "- [server_threaded.cc](./codec/server_threaded.cc)使用多线程`TcpServer，`并用`mutex`来保护共享数据。\n",
    "- [server_threaded_efficient.cc](./codec/server_threaded_efficient.cc)对共享数据以§ 2.8“借`shared_ptr`实现copy-on-write”的手法来降低锁竞争。\n",
    "- [server_threaded_highperformance.cc](./codec/server_threaded_highperformance.cc)采用`thread local`变量，实现多线程高效转发，这个例子值得仔细阅读理解。\n",
    "\n",
    "§ 7.8会介绍muduo中的定时器，并实现Boost.Asio教程中的timer2～5示例，以及带流量统计功能的discard和echo服务器（来自Java Netty）​。流量等于单位时间内发送或接收的字节数，这要用到定时器功能。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# * 7.4 muduo Buffer类的设计与使用"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "本节介绍muduo中输入输出缓冲区的设计与实现。文中buffer指一般的应用层缓冲区、缓冲技术，`Buffer`特指`muduo::net::Buffer class`。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## muduo的IO模型\n",
    "\n",
    "[UNP]6.2节总结了Unix/Linux上的五种IO模型：阻塞(blocking)、非阻塞(non-blocking)、IO复用(IO multiplexing)、信号驱动(signal-driven)、异步(asynchronous)。这些都是单线程下的IO模型。\n",
    "\n",
    "C10k问题[插图]的页面介绍了五种IO策略，把线程也纳入考量。​（现在C10k已经不是什么问题，C100k也不是大问题，C1000k才算得上挑战）​。\n",
    "> http://www.kegel.com/c10k.html\n",
    "\n",
    "\n",
    "在这个多核时代，线程是不可避免的。那么服务端网络编程该如何选择线程模型呢？我赞同libev作者的观点：[one loop per thread is usually a good model](http://pod.tst.eu/http://cvs.schmorp.de/libev/ev.pod#THREADS_AND_COROUTINES)。之前我也不止一次表述过这个观点，参见§ 3.3“多线程服务器的常用编程模型”和§ 6.6“详解muduo多线程模型”​。\n",
    "\n",
    "如果采用one loop per thread的模型，多线程服务端编程的问题就简化为如何设计一个高效且易于使用的event loop，然后每个线程run一个event loop就行了（当然、同步和互斥是不可或缺的）​。在“高效”这方面已经有了很多成熟的范例(libev、libevent、memcached、redis、lighttpd、nginx)，在“易于使用”方面我希望muduo能有所作为。​（muduo可算是用现代C++实现了Reactor模式，比起原始的Reactor来说要好用得多。​）\n",
    "\n",
    "event loop是non-blocking网络编程的核心，在现实生活中，non-blocking几乎总是和IO multiplexing一起使用，原因有两点：\n",
    "- 没有人真的会用轮询(busy-pooling)来检查某个non-blocking IO操作是否完成，这样太浪费CPU cycles。\n",
    "- IO multiplexing一般不能和blocking IO用在一起，因为blocking IO中`read()/write()/accept()/connect()`都有可能阻塞当前线程，这样线程就没办法处理其他socket上的IO事件了。见[UNP]16.6节“nonblocking accept”的例子。\n",
    "\n",
    "所以，当我提到non-blocking的时候，实际上指的是*non-blocking + IO multi-plexing*，单用其中任何一个是不现实的。另外，本书所有的“连接”均指TCP连接，socket和connection在文中可互换使用。\n",
    "\n",
    "当然，non-blocking编程比blocking难得多，见§ 6.4.1“TCP网络编程本质论”列举的难点。基于event loop的网络编程跟直接用C/C++编写单线程Windows程序颇为相像：程序不能阻塞，否则窗口就失去响应了；在`event handler`中，程序要尽快交出控制权，返回窗口的事件循环。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 为什么non-blocking网络编程中应用层buffer是必需的\n",
    "\n",
    "non-blocking IO的核心思想是避免阻塞在`read()`或`write()`或其他IO系统调用上，这样可以最大限度地复用thread-of-control，让一个线程能服务于多个socket连接。IO线程只能阻塞在IO multiplexing函数上，如`select/poll/epoll_wait`。这样一来，应用层的缓冲是必需的，每个TCP socket都要有stateful的input buffer和output buffer。\n",
    "\n",
    "TcpConnection必须要有output buffer 考虑一个常见场景：程序想通过TCP连接发送100kB的数据，但是在`write()`调用中，操作系统只接受了80kB（受TCP advertised window的控制，细节见[TCPv1]​）​，你肯定不想在原地等待，因为不知道会等多久（取决于对方什么时候接收数据，然后滑动TCP窗口）​。程序应该尽快交出控制权，返回event loop。在这种情况下，剩余的20kB数据怎么办？\n",
    "\n",
    "对于应用程序而言，它只管生成数据，它不应该关心到底数据是一次性发送还是分成几次发送，这些应该由网络库来操心，程序只要调用`TcpConnection::send()`就行了，网络库会负责到底。网络库应该接管这剩余的20kB数据，把它保存在该TCP connection的output buffer里，然后注册`POLLOUT`事件，一旦socket变得可写就立刻发送数据。当然，这第二次`write()`也不一定能完全写入20kB，如果还有剩余，网络库应该继续关注`POLLOUT`事件；如果写完了20kB，网络库应该停止关注`POLLOUT`，以免造成busy loop。​（muduo EventLoop采用的是epoll level trigger，原因见下页。​）\n",
    "\n",
    "如果程序又写入了50kB，而这时候output buffer里还有待发送的20kB数据，那么网络库不应该直接调用`write()`，而应该把这50kB数据append在那20kB数据之后，等socket变得可写的时候再一并写入。\n",
    "\n",
    "如果output buffer里还有待发送的数据，而程序又想关闭连接（对程序而言，调用`TcpConnection::send()`之后他就认为数据迟早会发出去）​，那么这时候网络库不能立刻关闭连接，而要等数据发送完毕，见p.191“为什么`TcpConnection::shutdown()`没有直接关闭TCP连接”中的讲解。\n",
    "\n",
    "`TcpConnection`必须要有input buffer, TCP是一个无边界的字节流协议，**接收方必须要处理“收到的数据尚不构成一条完整的消息”和“一次收到两条消息的数据”等情况**。一个常见的场景是，发送方`send()`了两条1kB的消息（共2kB）​，接收方收到数据的情况可能是：\n",
    "- 一次性收到 2 kB 数据；\n",
    "- 分两次收到：第一次 600 B，第二次 1400 B；\n",
    "- 分两次收到：第一次 1400 B，第二次 600 B；\n",
    "- 分两次收到：第一次 1 kB，第二次 1 kB；\n",
    "- 分三次收到：第一次 600 B，第二次 800 B，第三次 600 B；\n",
    "- 其他任意可能。一般而言，长度为 n 字节的消息分块到达的可能性有 $2^{n-1}$种。\n",
    "\n",
    "> AI: 非阻塞 IO 线程只能在 `select/poll/epoll_wait` 上阻塞，`read/write` 可能随时“读不够/写不完”，因此必须用应用层缓冲区来保存未处理的数据。\n",
    "\n",
    "网络库在处理“socket可读”事件的时候，必须一次性把socket里的数据读完（从操作系统buffer搬到应用层buffer）​，否则会反复触发`POLLIN`事件，造成busy-loop。那么网络库必然要应对“数据不完整”的情况，收到的数据先放到input buffer里，等构成一条完整的消息再通知程序的业务逻辑。这通常是`codec`的职责，见§7.3“Boost.Asio的聊天服务器”中的“TCP分包”的论述与代码。所以，在TCP网络编程中，网络库必须要给每个TCP connection配置input buffer。\n",
    "\n",
    "muduo EventLoop采用的是`epoll(4)`,level trigger，而不是edge trigger。一是为了与传统的`poll(2)`兼容，因为在文件描述符数目较少，活动文件描述符比例较高时，`epoll(4)`不见得比`poll(2)`更高效，必要时可以在进程启动时切换Poller。二是level trigger编程更容易，以往`select(2)/poll(2)`的经验都可以继续用，不可能发生漏掉事件的bug。三是读写的时候不必等候出现`EAGAIN`，可以节省系统调用次数，降低延迟。\n",
    "\n",
    "所有muduo中的IO都是带缓冲的IO(buffered IO)，你不会自己去`read()`或`write()`某个socket，只会操作`TcpConnection`的input buffer和output buffer。更确切地说，是在`onMessage()`回调里读取input buffer；调用`TcpConnection::send()`来间接操作output buffer，一般不会直接操作output buffer。\n",
    "\n",
    "> AI:\n",
    "> - **Output buffer**：写调用可能只发出部分数据（受 TCP 窗口等限制），剩余数据需要暂存并在 socket 变为可写时继续发送；发送完再取消关注 `POLLOUT`，也能保障关闭前数据全部送达。\n",
    "> - **Input buffer**：TCP 是无边界字节流，消息可能被拆分或粘连，网络库需把所有可读数据先拉到缓冲里，再根据协议解析完整消息，避免重复触发 `POLLIN`。\n",
    "\n",
    "另外，muduo的`onMessage()`的原型如下，它既可以是free function，也可以是member function，反正muduo `TcpConnection`只认`boost::function<>`。\n",
    "\n",
    "```c++\n",
    "void onMessage(const TcpConnectionPtr&,Buffer*, Timestamp)；\n",
    "```\n",
    "\n",
    "对于网络程序来说，一个简单的验收测试是：输入数据每次收到一个字节（200字节的输入数据会分200次收到，每次间隔10ms）​，程序的功能不受影响。对于muduo程序，通常可以用codec来分离“消息接收”与“消息处理”​，见§ 7.6“在muduo中实现Protobuf编解码器与消息分发器”对“编解码器codec”的介绍。\n",
    "\n",
    "如果某个网络库只提供相当于`char buf[8192]`的缓冲，或者根本不提供缓冲区，而仅仅通知程序“某socket可读/某socket可写”​，要程序自己操心IO buffering，这样的网络库用起来就很不方便了。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Buffer的功能需求\n",
    "\n",
    "muduo Buffer的设计考虑了常见的网络编程需求，我试图在易用性和性能之间找一个平衡点，目前这个平衡点更偏向于易用性。\n",
    "\n",
    "muduo Buffer的设计要点：\n",
    "- 对外表现为一块连续的内存（`char* p, int len`），以方便客户代码的编写。\n",
    "- `size()` 可以自动增长，以适应不同大小的消息；它不是一个 fixed size array（例如 `char buf[8192]`）。\n",
    "- 内部以 `std::vector<char>` 来保存数据，并提供相应的访问函数。\n",
    "\n",
    "\n",
    "Buffer 其实像是一个 queue，从末尾写入数据，从头部读出数据。\n",
    "\n",
    "谁会用 Buffer？谁写谁读？根据前文分析，`TcpConnection` 会有两个 Buffer 成员：input buffer 与 output buffer。\n",
    "- **input buffer**：`TcpConnection` 会从 socket 读取数据，然后写入 input buffer（其实这一步是用 `Buffer::readFd()` 完成的）；客户代码从 input buffer 读取数据。\n",
    "- **output buffer**：客户代码会把数据写入 output buffer（其实这一步是用 `TcpConnection::send()` 完成的）；`TcpConnection` 从 output buffer 读取数据并写入 socket。\n",
    "\n",
    "其实，input和output是针对客户代码而言的，客户代码从input读，往output写。`TcpConnection`的读写正好相反。\n",
    "\n",
    "图7-2是`muduo::net::Buffer`的类图。请注意，为了后面画图方便，这个类图跟实际代码略有出入，但不影响我要表达的观点。代码位于[muduo/net/Buffer.h](../muduo/muduo/net/Buffer.h)。\n",
    "\n",
    "| 可见性 | 成员/方法                   | 类型 / 签名         |\n",
    "| :----: | --------------------------- | ------------------- |\n",
    "|   −    | `data`                      | `std::vector<char>` |\n",
    "|   −    | `readIndex`                 | `int`               |\n",
    "|   −    | `writeIndex`                | `int`               |\n",
    "|   +    | `readableBytes()`           | `int`               |\n",
    "|   +    | `peek()`                    | `const char*`       |\n",
    "|   +    | `retrieve(int)`             | `void`              |\n",
    "|   +    | `retrieveAsString()`        | `std::string`       |\n",
    "|   +    | `append(const void*, int)`  | `void`              |\n",
    "|   +    | `prepend(const void*, int)` | `void`              |\n",
    "|   +    | `swap(Buffer&)`             | `void`              |\n",
    "|   −    | `readFd(int)`               | `int`               |\n",
    "\n",
    "本节不介绍每个成员函数的使用，而会详细讲解`readIndex`和`writeIndex`的作用。\n",
    "\n",
    "`Buffer::readFd()`我在p.138写道：\n",
    "> 在非阻塞网络编程中，如何设计并使用缓冲区？一方面我们希望减少系统调用，一次读的数据越多越划算，那么似乎应该准备一个大的缓冲区。另一方面希望减少内存占用。如果有10000个并发连接，每个连接一建立就分配各50kB的读写缓冲区的话，将占用1GB内存，而大多数时候这些缓冲区的使用率很低。muduo用`readv(2)`结合栈上空间巧妙地解决了这个问题。\n",
    "\n",
    "具体做法是，在栈上准备一个65536字节的`extrabuf`，然后利用`readv()`来读取数据，`iovec`有两块，第一块指向muduo Buffer中的writable字节，另一块指向栈上的`extrabuf`。这样如果读入的数据不多，那么全部都读到`Buffer`中去了；如果长度超过Buffer的writable字节数，就会读到栈上的`extrabuf`里，然后程序再把`extrabuf`里的数据`append()`到`Buffer`中，代码见§8.7.2。\n",
    "\n",
    "这么做利用了临时栈上空间，避免每个连接的初始Buffer过大造成的内存浪费，也避免反复调用`read()`的系统开销（由于缓冲区足够大，通常一次`readv()`系统调用就能读完全部数据）​。由于muduo的事件触发采用level trigger，因此这个函数并不会反复调用`read()`直到其返回`EAGAIN`，从而可以降低消息处理的延迟。\n",
    "> `readFd()`是最内层函数，其在每个IO线程的最大stack空间开销是固定的64KiB，与连接数目无关。如果stack空间紧张，也可以改用thread local的`extrabuf`，但是不能全局共享一个`extrabuf`。（为什么？）\n",
    "\n",
    "这算是一个小小的创新吧。\n",
    "\n",
    "线程安全？`muduo::net::Buffer`不是线程安全的（其安全性跟`std::vector`相同）​，这么设计的理由如下：\n",
    "- 对于 input buffer，`onMessage()` 回调始终发生在该 `TcpConnection` 所属的 IO 线程，应用程序应该在 `onMessage()` 完成对 input buffer 的操作，并且不要把 input buffer 暴露给其他线程。这样所有对 input buffer 的操作都在同一个线程，`Buffer` 类不必是线程安全的。\n",
    "- 对于 output buffer，应用程序不会直接操作它，而是调用 `TcpConnection::send()` 来发送数据，后者是线程安全的。\n",
    "\n",
    "代码中用 `EventLoop::assertInLoopThread()` 保证以上假设成立。\n",
    "\n",
    "如果 `TcpConnection::send()` 调用发生在该 `TcpConnection` 所属的那个 IO 线程，那么它会转而调用 `TcpConnection::sendInLoop()`，`sendInLoop()` 会在当前线程（也就是 IO 线程）操作 output buffer；如果 `TcpConnection::send()` 调用发生在别的线程，它不会在当前线程调用 `sendInLoop()`，而是通过 `EventLoop::runInLoop()` 把 `sendInLoop()` 函数调用转移到 IO 线程（听上去颇为神奇？），这样 `sendInLoop()` 还是会在 IO 线程操作 output buffer，不会有线程安全问题。当然，跨线程的函数转移调用涉及函数参数的跨线程传递，一种简单的做法是把数据拷贝一份，绝对安全。\n",
    "\n",
    "另一种更为高效的做法是用 `swap()`。这就是为什么 `TcpConnection::send()` 的某个重载以 `Buffer*` 为参数，而不是 `const Buffer&`，这样可以避免拷贝，而用 `Buffer::swap()` 实现高效的线程间数据转移。（最后这点，仅为设想，暂未实现。目前仍然以数据拷贝方式在线程间传递，略微有些性能损失。）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**核心要点概述**\n",
    "\n",
    "1. **设计目标**：muduo 的 `Buffer` 以易用性为优先，外部表现为连续内存（`char* + len`），内部用 `std::vector<char>` 存储，容量可自动扩展。\n",
    "2. **读写模型**：`Buffer` 类似队列，尾部写、头部读。`TcpConnection` 拥有输入与输出两个 `Buffer`：\n",
    "   - 输入缓冲：socket → `Buffer` → 用户代码读取。\n",
    "   - 输出缓冲：用户代码写入 → `Buffer` → socket 发送。\n",
    "3. **关键索引**：通过 `readIndex` 与 `writeIndex` 管理可读、可写区域；常用接口包括 `readableBytes()`、`peek()`、`append()`、`retrieve()`、`swap()` 等。\n",
    "4. **高效读入**：`readFd()` 使用 `readv` 结合固定 64KiB 栈上 `extrabuf`，在不开大初始缓冲区的情况下尽量减少系统调用与内存浪费；若数据超过当前可写空间，先写入 `extrabuf` 再追加入 `Buffer`。\n",
    "5. **线程安全策略**：`Buffer` 本身非线程安全，依赖事件循环保证：\n",
    "   - 输入缓冲仅在所属 IO 线程的 `onMessage()` 中操作。\n",
    "   - 输出缓冲通过线程安全的 `TcpConnection::send()` 间接访问，必要时使用 `EventLoop::runInLoop()` 将操作转移回 IO 线程。\n",
    "6. **跨线程优化设想**：`TcpConnection::send(Buffer*)` 重载可通过 `Buffer::swap()` 避免数据拷贝，实现高效线程间数据传递（目前尚未实现）。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Buffer的数据结构\n",
    "\n",
    "Buffer 的内部是一个 `std::vector<char>`，它是一块连续的内存。此外，`Buffer` 有两个 data member，指向该 `vector` 中的元素。这两个 `index` 的类型是 `int`，不是 `char*`，目的是应对迭代器失效。muduo `Buffer` 的设计参考了 Netty 的 `ChannelBuffer` 和 libevent 1.4.x 的 `evbuffer`。不过，其 `prependable` 可算是一点“微创新”。\n",
    "\n",
    "在介绍 `Buffer` 的数据结构之前，先简单说一下后面示意图中表示指针或下标的箭头所指位置的具体含义。对于长度为 10 的字符串 `\"Chen Shuo\\n\"`，如果 `p0` 指向第 0 个字符（白色区域的开始），`p1` 指向第 5 个字符（灰色区域的开始），`p2` 指向 `'\\n'` 之后的那个位置（通常是 `end()` 迭代器所指的位置），那么精确的画法如图 7-3 的左图所示，简略的画法如图 7-3 的右图所示，后文都采用这种简略画法。\n",
    "\n",
    "```txt\n",
    "       0    1    2    3    4    5    6    7    8    9   10\n",
    "     +----+----+----+----+----+----+----+----+----+----+\n",
    "     | C  | h  | e  | n  | _  | S  | h  | u  | o  | \\n |\n",
    "     +----+----+----+----+----+----+----+----+----+----+\n",
    "     ^                         ^                         ^\n",
    "     |                         |                         |\n",
    "    p0                        p1                        p2\n",
    "\n",
    "       0    1    2    3    4    5    6    7    8    9   10\n",
    "     +----+----+----+----+----+----+----+----+----+----+\n",
    "     |    |    |    |    |    |####|####|####|####|####|\n",
    "     +----+----+----+----+----+----+----+----+----+----+\n",
    "     ^                         ^                         ^\n",
    "     |                         |                         |\n",
    "    p0                        p1                        p2\n",
    "```\n",
    "\n",
    "<img src=\"./images/7.3.png\" alt=\"7.3\" style=\"zoom:50%;\" />\n",
    "\n",
    "muduo Buffer的数据结构如图7-4所示。\n",
    "\n",
    "<img src=\"./images/7.4.png\" alt=\"7.4\" style=\"zoom:50%;\" />\n",
    "\n",
    "两个index把`vector`的内容分为三块：`prependable`、`readable`、`writable`，各块的大小见式7−1。灰色部分是Buffer的有效载荷(payload)，`prependable`的作用留到后面讨论。\n",
    "```txt\n",
    "prependable = readIndex\n",
    "readable    = writeIndex - readIndex\n",
    "write       = size() - writeIndex\n",
    "```\n",
    "\n",
    "readIndex和writeIndex满足以下**不变式(invariant)**：$0 \\leq \\text{readIndex} \\leq \\text{writeIndex} \\leq \\text{size}()$\n",
    "\n",
    "\n",
    "muduo Buffer里有两个常数`kCheapPrepend`和`kInitialSize`，定义了`prependable`的初始大小和`writable`的初始大小，`readable`的初始大小为`0`。在初始化之后，Buffer的数据结构如图7-5所示，其中括号里的数字是该变量或常量的值。\n",
    "\n",
    "<img src=\"./images/7.5.png\" alt=\"7.5\" style=\"zoom:50%;\" />\n",
    "\n",
    "根据以上公式（见式7−1）可算出各块的大小，刚刚初始化的Buffer里没有payload数据，所以`readable = 0`。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Buffer的操作"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 基本的read-write cycle\n",
    "\n",
    "Buffer初始化后的情况见图7-4。如果向Buffer写入了200字节，那么其布局如图7-6所示。\n",
    "\n",
    "<img src=\"./images/7.6.png\" alt=\"7.6\" style=\"zoom:50%;\" />\n",
    "\n",
    "图7-6中`writeIndex`向后移动了200字节，`readIndex`保持不变，`readable`和`writable`的值也有变化。\n",
    "\n",
    "如果从`Buffer read()&retrieve()`（下称“读入”）了50字节，结果如图7-7所示。与图7-6相比，`readIndex`向后移动50字节，`writeIndex`保持不变，`readable`和`writable`的值也有变化（这句话往后从略）。\n",
    "\n",
    "<img src=\"./images/7.7.png\" alt=\"7.7\" style=\"zoom:50%;\" />\n",
    "\n",
    "然后又写入了200字节，`writeIndex`向后移动了200字节，`readIndex`保持不变，如图7-8所示。\n",
    "\n",
    "<img src=\"./images/7.8.png\" alt=\"7.8\" style=\"zoom:50%;\" />\n",
    "\n",
    "接下来，一次性读入350字节，请注意，由于全部数据读完了，`readIndex`和`writeIndex`返回原位以备新一轮使用（见图7-9）​，这和图7-5是一样的。\n",
    "\n",
    "<img src=\"./images/7.9.png\" alt=\"7.9\" style=\"zoom:50%;\" />\n",
    "\n",
    "以上过程可以看作是发送方发送了两条消息，长度分别为50字节和350字节，接收方分两次收到数据，每次200字节，然后进行分包，再分两次回调客户代码。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 自动增长\n",
    "\n",
    "muduo Buffer不是固定长度的，它可以自动增长，这是使用`vector`的直接好处。假设当前的状态如图7-10所示。​（这和前面的图7-8是一样的。​）\n",
    "\n",
    "<img src=\"./images/7.10.png\" alt=\"7.10\" style=\"zoom:50%;\" />\n",
    "\n",
    "客户代码一次性写入1000字节，而当前可写的字节数只有624，那么buffer会自动增长以容纳全部数据，得到的结果如图7-11所示。注意`readIndex`返回到了前面，以保持`prependable`等于`kCheapPrependable`。由于`vector`重新分配了内存，原来指向其元素的指针会失效，这就是为什么`readIndex`和`writeIndex`是整数下标而不是指针。​（注意：在目前的实现中`prependable`会保持58字节，留待将来修正。​）\n",
    "\n",
    "<img src=\"./images/7.11.png\" alt=\"7.11\" style=\"zoom:50%;\" />\n",
    "\n",
    "然后读入350字节，`readIndex`前移，如图7-12所示。\n",
    "\n",
    "<img src=\"./images/7.12.png\" alt=\"7.12\" style=\"zoom:50%;\" />\n",
    "\n",
    "最后，读完剩下的1000字节，`readIndex`和`writeIndex`返回`kCheapPrependable`，如图7-13所示。\n",
    "\n",
    "<img src=\"./images/7.13.png\" alt=\"7.13\" style=\"zoom:50%;\" />\n",
    "\n",
    "注意`buffer`并没有缩小大小，下次写入1350字节就不会重新分配内存了。换句话说，`muduo Buffer`的`size()`是自适应的，它一开始的初始值是1kB多，如果程序中经常收发10kB的数据，那么用几次之后它的`size()`会自动增长到10kB，然后就保持不变。这样一方面避免浪费内存（`Buffer`的初始大小直接决定了高并发连接时的内存消耗），另一方面避免反复分配内存。当然，客户代码可以手动`shrink() buffer size()`。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### `size()`与`capacity()`\n",
    "\n",
    "使用`vector`的另一个好处是它的`capcity()`机制减少了内存分配的次数。比方说程序反复写入1字节，`muduo Buffer`不会每次都分配内存，`vector`的`capacity()`以指数方式增长，让`push_back()`的平均复杂度是常数。比方说经过第一次增长，`size()`刚好满足写入的需求，如图7-14所示。但这个时候`vector`的`capacity()`已经大于`size()`，在接下来写入`capacity()−size()`字节的数据时，都不会重新分配内存，如图7-15所示。\n",
    "\n",
    "<img src=\"./images/7.14.png\" alt=\"7.14\" style=\"zoom:50%;\" />\n",
    "\n",
    "<img src=\"./images/7.15.png\" alt=\"7.15\" style=\"zoom:50%;\" />\n",
    "\n",
    "思考题：为什么我们不需要调用 `reserve()` 来预先分配空间？因为 `Buffer` 在构造函数里把初始 `size()` 设为 1 KiB，这样当 `size()` 超过 1 KiB 的时候 `vector` 会把 `capacity()` 加倍，等于说 `resize()` 替我们做了 `reserve()` 的事。用一段简单的代码验证一下："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0 0\n",
      "1024 1024\n",
      "1300 2048\n"
     ]
    }
   ],
   "source": [
    "#include <cstdio>\n",
    "#include <vector>\n",
    "\n",
    "{\n",
    "    std::vector<char> vec;\n",
    "    std::printf(\"%zu %zu\\n\", vec.size(), vec.capacity()); // 一开始 size() 和 capacity() 都是 0\n",
    "\n",
    "    vec.resize(1024);\n",
    "    std::printf(\"%zu %zu\\n\", vec.size(), vec.capacity()); // resize(1024) 之后 size() 和 capacity() 都是 1024\n",
    "\n",
    "    vec.resize(1300);\n",
    "    std::printf(\"%zu %zu\\n\", vec.size(), vec.capacity()); // esize(稍大) 之后 capacity() 翻倍，相当于 reserve(2048)\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "细心的读者可能会发现用 `capacity()` 也不是完美的，它有优化的余地。具体来说，`vector::resize()` 会初始化（`memset()`/`bzero()`）内存，而我们不需要它初始化，因为反正立刻就要填入数据。比如，在图7-15的基础上写入200字节，由于 `capacity()` 足够大，不会重新分配内存，这是好事；但是 `vector::resize()` 会先把那200字节 `memset()` 为`0`（见图7-16），然后 `muduo Buffer` 再填入数据（见图7-17）。这么做稍微有点浪费，不过我不打算优化它，除非它确实造成了性能瓶颈。（精通STL的读者可能会说用 `vec.insert`（`vec.end()`，…）以避免浪费，但是 `writeIndex` 和 `size()` 不一定是对齐的，会有别的麻烦。）\n",
    "\n",
    "<img src=\"./images/7.16.png\" alt=\"7.16\" style=\"zoom:50%;\" />\n",
    "\n",
    "<img src=\"./images/7.17.png\" alt=\"7.17\" style=\"zoom:50%;\" />\n",
    "\n",
    "Google Protobuf中有一个[STLStringResizeUninitialized](https://github.com/protocolbuffers/protobuf/blob/main/src/google/protobuf/io/zero_copy_stream_impl_lite.cc#L148)函数，干的就是这个事情。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 内部腾挪\n",
    "\n",
    "有时候，经过若干次读写，`readIndex`移到了比较靠后的位置，留下了巨大的`prependable`空间，如图7-18所示。\n",
    "\n",
    "<img src=\"./images/7.18.png\" alt=\"7.18\" style=\"zoom:50%;\" />\n",
    "\n",
    "这时候，如果我们想写入300字节，而`writable`只有200字节，怎么办？muduo Buffer在这种情况下不会重新分配内存，而是先把已有的数据移到前面去，腾出`writable`空间，如图7-19所示。\n",
    "\n",
    "<img src=\"./images/7.19.png\" alt=\"7.19\" style=\"zoom:50%;\" />\n",
    "\n",
    "然后，就可以写入300字节了，如图7-20所示。\n",
    "\n",
    "<img src=\"./images/7.20.png\" alt=\"7.20\" style=\"zoom:50%;\" />\n",
    "\n",
    "这么做的原因是，如果重新分配内存，反正也是要把数据拷贝到新分配的内存区域，代价只会更大"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 前方添加(prepend)\n",
    "\n",
    "前面说muduo Buffer有个小小的创新（或许不是创新，我记得在哪儿看到过类似的做法，忘了出处）​，即提供prependable空间，让程序能以很低的代价在数据前面添加几个字节。\n",
    "\n",
    "比方说，程序以固定的4个字节表示消息的长度（§ 7.3“Boost.Asio的聊天服务器”中的`LengthHeaderCodec`）​，我要序列化一个消息，但是不知道它有多长，那么我可以一直`append()`直到序列化完成（图7-21，写入了200字节）​，然后再在序列化数据的前面添加消息的长度（图7-22，把200这个数`prepend`到首部）​。\n",
    "\n",
    "<img src=\"./images/7.21.png\" alt=\"7.21\" style=\"zoom:50%;\" />\n",
    "\n",
    "<img src=\"./images/7.22.png\" alt=\"7.22\" style=\"zoom:50%;\" />\n",
    "\n",
    "通过预留`kCheapPrependable`空间，可以简化客户代码，以空间换时间\n",
    "\n",
    "以上各种use case的单元测试见[muduo/net/tests/Buffer_unittest.cc](../muduo/muduo/net/tests/Buffer_unittest.cc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 其他设计方案\n",
    "\n",
    "这里简单谈谈其他可能的应用层buffer设计方案。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 不用vector<char>\n",
    "\n",
    "如果有STL洁癖，那么可以自己管理内存，以4个指针为`buffer`的成员，数据结构如图7-23所示。\n",
    "\n",
    "<img src=\"./images/7.23.png\" alt=\"7.23\" style=\"zoom:50%;\" />\n",
    "\n",
    "说实话我不觉得这种方案比`std::vector`好。代码变复杂了，性能也未见得有能察觉得到(noticeable)的改观。如果放弃“连续性”要求，可以用circular buffer，这样可以减少一点内存拷贝（没有“内部腾挪”​）​。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### zero copy\n",
    "\n",
    "如果对性能有极高的要求，受不了`copy()`与`resize()`，那么可以考虑实现分段连续的zero copy buffer再配合gather scatter IO，数据结构如图7-24所示，这是libevent 2.0.x的设计方案。TCPv2介绍的BSD TCP/IP实现中的`mbuf`也是类似的方案，Linux的`sk_buff`估计也差不多。细节有出入，但基本思路都是不要求数据在内存中连续，而是用链表把数据块链接到一起。\n",
    "\n",
    "<img src=\"./images/7.24.png\" alt=\"7.24\" style=\"zoom:50%;\" />\n",
    "\n",
    "图 7‑24 绘制的是由两个 `evbuffer_chain` 构成的 `evbuffer`，右边两个 `evbuffer_chain` 结构体中深灰色的部分是 payload，可见 `evbuffer` 的缓冲区不是连续的，而是分块的。\n",
    "\n",
    "当然，高性能的代价是代码变得晦涩难读，buffer 不再是连续的，parse 消息会稍微麻烦一些。如果你的程序只处理 `Protobuf Message`，这不是问题，因为 Protobuf 有 `ZeroCopyInputStream` 接口，只要实现这个接口，parsing 的事情就交给 `Protobuf Message` 去操心了。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 性能是不是问题\n",
    "\n",
    "看到这里，有的读者可能会嘀咕：muduo Buffer 有那么多可以优化的地方，其性能会不会太低？对此，我的回应是“可以优化，不一定值得优化。”\n",
    "\n",
    "muduo 的设计目标是用于开发公司内部的分布式程序。换句话说，它是用来写专用的 Sudoku server 或者游戏服务器，不是用来写通用的 httpd 或 ftpd 或 Web proxy。前者通常有业务逻辑，后者更强调高并发与高吞吐量。\n",
    "\n",
    "以 Sudoku 为例，假设求解一个 Sudoku 问题需要 0.2 ms，服务器有 8 个核，那么理想情况下每秒最多能求解 40000 个问题。每次 Sudoku 请求的数据大小低于 100 字节（一个 9×9 的数独只要 81 字节，加上 header 也可以控制在 100 字节以下），也就是说 100×40000=4 MB/s 的吞吐量就足以让服务器的 CPU 饱和。在这种情况下，去优化 Buffer 的内存拷贝次数似乎没有意义。，如果你写一个实实在在有业务的服务器程序（比如分布式存储、搜索、微博、短网地址），由于市面上没有可等的前任，也不需在优化上投入全部精力，只要一个版本做得比一个版本好就行。先正确实现所需的功能，较少产生问题，然后再根据真实的负载情况来做优化，恐怕比在编码阶段就盲目优化要更 effective 一些。\n",
    "\n",
    "再举一个例子，目前最常用的千兆以太网的裸吞吐量是 125 MB/s，扣除以太网 header、IP header、TCP header 之后，应用层的吞吐率大约在 117 MB/s 左右。而现在服务器上最常用的 DDR2/DDR3 内存的带宽至少是 4 GB/s，比千兆以太网高 40 倍以上。也就是说，对于几十 kB 或几十 kB 大小的数据，在内存中复制几次根本不是问题，因为受千兆以太网延迟和带宽的限制，跟这个程序通信的其他机器上的程序不会察觉到性能差异。\n",
    "\n",
    "> 在不考虑 jumbo frame 的情况下，估算过程是：对于千兆以太网，每秒能传输 1000 Mbit 数据，即 125000000 B/s，每个以太网 frame 的固定开销有：preamble(8 B)、MAC(12 B)、type(2 B)、payload(46 B～1500 B)、CRC(4 B)、gap(12 B)，因此最小的以太网帧是 84 B，每秒可发送约 1488000 帧（换言之，对于一问一答的 RPC，其 qps 上限约是 700 k/s），最大的以太网帧是 1538 B，每秒可发送 81274 帧。再来算 TCP 的表数据：一个 TCP segment 包含 IpHeader(20 B) 和 TCPHeader(20 B)，还有 Timestamp option(12 B)，因此 TCP 的最大吞吐量是 81274 × (1500 - 52) = 117 MB/s，合 112 MiB/s。实测则是 7.8 s。\n",
    "\n",
    "最后举一个例子，如果你实现的服务程序要跟数据库打交道，那么瓶颈常常在 DB 上，优化服务程序本身不见得能提高性能（DB 读一次数据往往就抵消了你做的全部 low-level 优化），这时不如把精力投入在 DB 调优上。\n",
    "\n",
    "专用服务程序与通用服务程序的另外一点区别是 benchmark 的对象不同。如果你打算写一个 httpd，自然有人会拿来和目前最好的 Nginx 对比，立马就能把出性能高低。而且，如果你写一个实现公司内部业务的服务程序（比如分布式存储、搜索、微博、短网址），由于市面上没有同等功能的开源实现，你不需要在优化上投入全部精力，只要一版做得比一版好就行。先正确实现所需的功能，投入生产应用，然后再根据真实的负载情况来优化，这恐怕比在编码阶段就盲目调优要更有效一些。\n",
    "\n",
    "muduo的设计目标之一是吞吐量能让千兆以太网饱和，也就是每秒收发120MB数据。这个很容易就达到，不用任何特别的努力。\n",
    "\n",
    "如果确实在内存带宽方面遇到问题，说明你做的应用实在太critical，或许应该考虑放到Linux kernel里边去，而不是在用户态尝试各种优化。毕竟只有把程序做到kernel里才能真正实现zero copy；否则，核心态和用户态之间始终是有一次内存拷贝的。如果放到kernel里还不能满足需求，那么要么自己写新的kernel，或者直接用FPGA或ASIC操作network adapter来实现你的“高性能服务器”​。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# * 7.5 一种自动反射消息类型的Protobuf网络传输方案"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "本节假定读者了解Google Protocol Buffers是什么，这不是一篇Protobuf入门教程。本节的示例代码位于examples/protobuf/codec。\n",
    "\n",
    "本节要解决的问题是：通信双方在编译时就共享proto文件的情况下，接收方在收到Protobuf二进制数据流之后，如何自动创建具体类型的`Protobuf Message`对象，并用收到的数据填充该`Message`对象（即反序列化）​。​“自动”的意思是：当程序中新增一个`Protobuf Message`类型时，这部分代码不需要修改，不需要自己去注册消息类型。其实，Google Protobuf本身具有很强的反射(reflection)功能，可以根据`type name`创建具体类型的`Message`对象，我们直接利用即可。\n",
    "> Protobuf C++ 库的反射能力不止于此，它可以在运行时读入并解析任意 proto 文件，然后分析其中对应的二进制数据。有兴趣的读者请参考王益的博客 http://cxwangyi.blogspot.com/2010/06/google-protocol-buffers-proto.html"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 网络编程中使用Protobuf的两个先决条件\n",
    "\n",
    "Google Protocol Buffers（简称Protobuf）是一款非常优秀的库，它定义了一种紧凑（compact，相对XML和JSON而言）的可扩展二进制消息格式，特别适合网络数据传输。\n",
    "\n",
    "它为多种语言提供binding，大大方便了分布式程序的开发，让系统不再局限于用某一种语言来编写。\n",
    "\n",
    "在网络编程中使用Protobuf需要解决以下两个问题。\n",
    "1. 长度，Protobuf打包的数据没有自带长度信息或终结符，需要由应用程序自己在发生和接收的时候做正确的切分。\n",
    "2. 类型，Protobuf打包的数据没有自带类型信息，需要由发送方把类型信息传给给接收方，接收方创建具体的`Protobuf Message`对象，再做反序列化。\n",
    "\n",
    "Protobuf这么设计的原因见下一节。\n",
    "- 这里第一个问题很好解决，通常的做法是在每个消息前面加个固定长度的length header，例如§ 7.3中实现的`LengthHeaderCodec`。\n",
    "- 第二个问题其实也很好解决，Protobuf对此有内建的支持。但是奇怪的是，从网上简单搜索的情况看，我发现了很多“山寨”的做法。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 山寨”做法\n",
    "\n",
    "以下均为在 Protobuf data 之前加上 header，header 中包含消息长度和类型信息。类型信息的“山寨”做法主要有两种：\n",
    "\n",
    "- 在 header 中放 `int typeId`，接收方用 switch-case 来选择对应的消息类型和处理函数；\n",
    "- 在 header 中放 `string typeName`，接收方用 look-up table 来选择对应的消息类型和处理函数。\n",
    "\n",
    "这两种做法都有问题。\n",
    "- 第一种做法要求保持 `typeId` 的唯一性，它和 `Protobuf message type` 一一对应。如果 `Protobuf message` 的使用范围不广，比如接收方和发送方都是自己维护的程序，那么 `typeId` 的唯一性不难保证，用版本管理工具即可。如果 `Protobuf message` 的使用范围很大，比如全公司都在用，而且不同部门开发的分布式程序可能相互通信，那么就需要一个公司内部的全局机构来分配 `typeId`，每次增加新 `message type` 都要去注册一下，比较麻烦。\n",
    "- 第二种做法稍好一点。`typeName` 的唯一性比较好办，因为可以加上 `package name`（也就是 `message` 的 fully qualified type name），各个部门事先分好 `namespace`，不会冲突与重复。但是每次新增消息类型的时候都要去修改 look-up table 的初始化代码，也比较麻烦。\n",
    "\n",
    "其实，不需要自己重新发明轮子，Protobuf 本身已经自带了解决方案。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 根据type name反射自动创建Message对象\n",
    "\n",
    "Google Protobuf 本身具有很强的反射（reflection）功能，可以根据 `type name` 创建具体类型的 `Message` 对象。但是奇怪的是，其官方教程里没有明确提及这个用法，我估计还有很多人不知道这个用法，所以觉得值得谈一谈。\n",
    "\n",
    "以下是笔者绘制的 Protobuf class diagram（见图 7-25）。我估计大家通常关心和使用的是这个类图的左半部分：`MessageLite`、`Message`、`Generated Message Types`（`Person`、`AddressBook`）等，而较少注意到图 7-25 的右半部分：`Descriptor`、`DescriptorPool`、`MessageFactory`。\n",
    "\n",
    "在图 7-25 中，起关键作用的是 `Descriptor` class，每个具体 `Message type` 对应一个 `Descriptor` 对象。尽管我们没有直接调用它的函数，但是 `Descriptor` 在“根据 `type name` 创建具体类型的 `Message` 对象”中扮演了重要的角色，起了桥梁作用。图 7-25 中的一箭头描述了根据 `type name` 创建具体 `Message` 对象的过程，后文会详细介绍。\n",
    "\n",
    "<img src=\"./images/7.25.png\" alt=\"7.25\" style=\"zoom:50%;\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 原理简述\n",
    "\n",
    "Protobuf `Message` class 采用了 [`Prototype pattern`](https://en.wikipedia.org/wiki/Prototype_pattern)，`Message` class 定义了 `New()` 虚函数，用以返回本对象的一份新实例，类型与本对象的真实类型相同。也就是说，拿到 `Message*` 指针，不用知道它的具体类型，就能创建和其类型一样的具体 `Message type` 的对象。\n",
    "\n",
    "每个具体 `Message type` 都有一个 `default instance`，可以通过 `ConcreteMessage::default_instance()` 获得，也可以通过 `MessageFactory::GetPrototype(const Descriptor*)` 来获得。所以，现在问题转变为：1. 如何拿到 `MessageFactory`；2. 如何拿到 `Descriptor*`。\n",
    "\n",
    "当然，`ConcreteMessage::descriptor()` 返回了我们想要的 `Descriptor*`，但是，在不知道 `ConcreteMessage` 的时候，如何调用它的静态成员函数呢？这似乎是个鸡与蛋的问题。\n",
    "\n",
    "我们的英雄是 `DescriptorPool`，它可以根据 `type name` 查到 `Descriptor*`，只要找到合适的 `DescriptorPool`，再调用 `DescriptorPool::FindMessageTypeByName(const string& type_name)` 即可。看到图 7-25 是不是眼前一亮？\n",
    "\n",
    "在最终解决问题之前，先简单测试一下，看看我上面说得对不对。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 验证思路\n",
    "\n",
    "本文用于举例的[proto](./protobuf/codec/query.proto)文件：\n",
    "\n",
    "其中的`Query.questioner`和`Answer.answerer`是§ 9.4提到的“分布式系统中的进程标识\n",
    "\n",
    "以下代码验证 `ConcreteMessage::default_instance()`、`ConcreteMessage::descriptor()`、`MessageFactory::GetPrototype()`、`DescriptorPool::FindMessageTypeByName()` 之间的不变式 (invariant)，注意其中的 `assert`：\n",
    "- [descriptor_test.cc](../recipes/protobuf/descriptor_test.cc)\n",
    "\n",
    "程序运行结果如下：\n",
    "```shell\n",
    "muduo.Query\n",
    "FindMessageTypeByName() = 0xd4e720\n",
    "T::descriptor() = 0xd4e720\n",
    "\n",
    "GetPrototype() = 0xd47710\n",
    "T::default_instance() = 0xd47710\n",
    "\n",
    "prototype->New() = 0xd459e0\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 根据`type name`自动创建`Message`的关键代码\n",
    "\n",
    "好了，万事俱备，开始行动：\n",
    "\n",
    "1. 用 `DescriptorPool::generated_pool()` 找到一个 `DescriptorPool` 对象，它包含了程序编译时所链接的全部 Protobuf `Message` types。\n",
    "2. 根据 `type name` 用 `DescriptorPool::FindMessageTypeByName()` 查找 `Descriptor`。\n",
    "3. 再用 `MessageFactory::generated_factory()` 找到 `MessageFactory` 对象，它能创建程序编译时所链接的全部 Protobuf `Message` types。\n",
    "4. 然后，用 `MessageFactory::GetPrototype()` 找到具体 `Message type` 的 `default instance`。\n",
    "5. 最后，用 `prototype->New()` 创建对象。\n",
    "\n",
    "示例代码如下。\n",
    "- [code.cc](./protobuf/codec/codec.cc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 调用方式：\n",
    "\n",
    "Message* newQuery = createMessage(\"muduo.Query\");\n",
    "assert(newQuery != NULL);\n",
    "assert(typeid(*newQuery) == typeid(muduo::Query::default_instance()));\n",
    "cout << \"createMessage(\\\"muduo.Query\\\") = \" << newQuery << endl;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "确实能从消息名称创建消息对象，古之人不余欺也 :-)\n",
    "\n",
    "注意，`createMessage()` 返回的是动态创建的对象的指针，调用方有责任释放它，不然就会使内存泄漏。在 muduo 里，我用 `shared_ptr<Message>` 来自动管理 Message 对象的生命周期。\n",
    "\n",
    "拿到 `Message*` 之后怎么办呢？怎么调用这个具体消息类型的处理函数？这就需要消息分发器 (dispatcher) 出马了，且听下回分解。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 线程安全性\n",
    "\n",
    "Google 的文档说，我们用到的那几个 `MessageFactory` 和 `DescriptorPool` 都是线程安全的，`Message::New()` 也是线程安全的，并且它们都是 const member function。关键问题解决了，那么剩下的工作就是设计一种包含长度和消息类型的 Protobuf 传输格式。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Protobuf传输格式\n",
    "\n",
    "笔者设计了一个简单的格式，包含Protobuf data和其对应的长度与类型信息，消息的末尾还有一个check sum。格式如图7-26所示，图中方块的宽度是32-bit。\n",
    "\n",
    "<img src=\"./images/7.26.png\" alt=\"7.26\" style=\"zoom:50%;\" />\n",
    "\n",
    "用C struct伪代码描述：\n",
    "```C++\n",
    "struct ProtobufTransportFormat __attribute__ ((__packed__))\n",
    "{\n",
    "  int32_t len;\n",
    "  int32_t nameLen;\n",
    "  char typeName[nameLen];\n",
    "  char protobufData[len - nameLen - 8];\n",
    "  int32_t checkSum;  // adler32 of nameLen, typeName and protobufData\n",
    "};\n",
    "```\n",
    "\n",
    "注意，这个格式不要求32-bit对齐，我们的decoder会自动处理非对齐的消息"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 例子\n",
    "\n",
    "用这个格式打包一个`muduo.Query`对象的结果如图7-27所示。\n",
    "\n",
    "<img src=\"./images/7.27.png\" alt=\"7.27\" style=\"zoom:50%;\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 设计决策\n",
    "\n",
    "以下是我在设计这个传输格式时的考虑：\n",
    "- signed int。消息中的长度字段只使用了 signed 32-bit int，而没有使用 unsigned int，这是为了跨语言移植性，因为 Java 语言没有 unsigned 类型。另外，Protobuf 一般用于打包小于 1 MB 的数据，unsigned int 也没用。\n",
    "- check sum。虽然 TCP 是可靠传输协议，虽然 Ethernet 有 CRC-32 校验，但是网络传输必须要考虑数据损坏的情况，对于关键的网络应用，check sum 是必不可少的。见 §A.1.13 “TCP 的可靠性有多高”。对于 Protobuf 这种紧凑的二进制格式而言，肉眼看不出数据有没有问题，需要用 check sum。\n",
    "- adler32 算法。我没有选用常见的 CRC-32，而是选用了 adler32，因为它的计算量小、速度比较快，强度和 CRC-32 差不多。另外，zlib 和 java.util.zip 都直接支持这个算法，不用我们自己实现。\n",
    "- type name 以 `\\0` 结束。这是为了方便 troubleshooting，比如通过 tcpdump 抓下来的包可以用肉眼很容易看出 type name，而不用根据 nameLen 去一个字节一个字节地数。同时，为了方便接收方处理，加入了 nameLen，节省了 `strlen()`，这是以空间换时间的做法。\n",
    "- 没有版本号。Protobuf Message 的一个突出优点是用 optional fields 来避免协议的版本号（凡是在 Protobuf Message 里放版本号的人都没有理解 Protobuf 的设计，甚至可能没有仔细阅读 Protobuf 的文档），让通信双方的程序能各自升级，便于系统演化。如果我设计的这个传输格式又把版本号加进去，那就画蛇添足了。\n",
    "> - http://code.google.com/apis/protocolbuffers/docs/overview.html “A bit of history”\n",
    "> - http://code.google.com/apis/protocolbuffers/docs/proto.html#updating\n",
    "> - http://code.google.com/apis/protocolbuffers/docs/cpptutorial.html “Extending a Protocol Buffer”\n",
    "\n",
    "Protobuf 可谓是网络协议格式的典范，值得我单独花一节篇幅讲述其思想，见 §9.6.1 “可扩展的消息格式”。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# * 7.6 在muduo中实现Protobuf编解码器与消息分发器\n",
    "\n",
    "本节是前一节的自然延续，介绍如何将前文介绍的打包方案与`muduo::net::Buffer`结合，实现Protobuf codec和dispatcher。在介绍codec和dispatcher之前，先讲讲前文的一个未决问题。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 为什么Protobuf的默认序列化格式没有包含消息的长度与类型\n",
    "\n",
    "Protobuf 是经过深思熟虑的消息打包方案，它的默认序列化格式没有包含消息的长度与类型，自然有其道理。哪些情况下不需要在 Protobuf 序列化得到的字节流中包含消息的长度和（或）类型？我能想到的答案有：\n",
    "- 如果把消息写入文件，一个文件存一个消息，那么序列化结果中不需要包含长度和类型，因为从文件名和文件长度中可以得知消息的类型与长度。\n",
    "- 如果把消息写入文件，一个文件存多个消息，那么序列化结果中不需要包含类型，因为文件名就代表了消息的类型。\n",
    "- 如果把消息存入数据库（或者 NoSQL），以 VARBINARY 字段保存，那么序列化结果中不需要包含长度和类型，因为从字段名和字段长度中可以得知消息的类型与长度。\n",
    "- 如果把消息以 UDP 方式发送给对方，而且对方一个 UDP port 只接收一种消息类型，那么序列化结果中不需要包含长度和类型，因为从 port 和 UDP packet 长度中可以得知消息的类型与长度。\n",
    "- 如果把消息以 TCP 短连接方式发给对方，而且对方一个 TCP port 只接收一种消息类型，那么序列化结果中不需要包含长度和类型，因为从 port 和 TCP 字节流长度中可以得知消息的类型与长度。\n",
    "- 如果把消息以 TCP 长连接方式发给对方，但是对方一个 TCP port 只接收一种消息类型，那么序列化结果中不需要包含类型，因为 port 代表了消息的类型。\n",
    "- 如果采用 RPC 方式通信，那么只需要告诉对方 `method name`，对方自然能推断出 `Request` 和 `Response` 的消息类型，这些可以由 protoc 生成的 RPC stubs 自动搞定。\n",
    "\n",
    "对于以上最后一点，比如说 `sudoku.proto` 的定义是：\n",
    "\n",
    "```proto\n",
    "service SudokuService {\n",
    "  rpc Solve (SudokuRequest) returns (SudokuResponse);\n",
    "}\n",
    "```\n",
    "\n",
    "那么 RPC method `SudokuService.Solve` 对应的请求和响应分别是 `SudokuRequest` 和 `SudokuResponse`。在发送 RPC 请求的时候，不需要包含 `SudokuRequest` 的类型，只需要发送 `method name SudokuService.Solve`，对方自然知道应该按 `SudokuRequest` 来解析 (parse) 请求。\n",
    "\n",
    "对于上述这些情况，如果 Protobuf 无条件地把长度和类型放到序列化的字节串中，只会浪费网络带宽和存储。可见 Protobuf 默认不发送长度和类型是正确的决定。Protobuf 为消息格式的设计树立了典范，哪些该自己搞定，哪些留给外部系统去解决，这些都考虑得很清楚。\n",
    "\n",
    "只有在使用 TCP 长连接，且在一个连接上传递不止一种消息的情况下（比方同时发 Heartbeat 和 Request/Response），才需要我前文提到的那种打包方案。这时候我们需要一个分发器 dispatcher，把不同类型的消息分给各个消息处理函数，这正是本节的主题之一。\n",
    "> 为什么要在一个连接上同时发 Heartbeat 和业务消息？见 § 9.3。\n",
    "\n",
    "以下均只考虑 TCP 长连接这一应用场景。先谈谈编解码器。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> - Protobuf 只负责序列化 `message` 内部字段，不自带消息长度和类型封装。\n",
    "> - 在许多传输/存储场景中，上层已经能提供长度或类型信息，重复添加只会浪费。\n",
    "> - 只有当多种消息共用同一长连接时，才需要由应用层额外定义“长度 + 类型”包头和分发逻辑"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 什么是编解码器(codec)\n",
    "\n",
    "[编解码器（codec）](http://en.wikipedia.org/wiki/Codec)是 encoder 和 decoder 的缩写，这是一个软硬件领域都在使用的术语，这里我借指“把网络数据和业务消息之间互相转换”的代码。\n",
    "\n",
    "在最简单的网络编程中，没有消息（message），只有字节流数据，这时候是用不到 codec 的。比如我们前面讲过的 echo server，它只需要把收到的数据原封不动地发送回去，而不必关心消息的边界（也没有“消息”的概念），收多少就发多少，这种情况下它干脆直接使用 `muduo::net::Buffer`，取到数据再交给 `TcpConnection` 发送回去，如图 7-28 所示。\n",
    "\n",
    "<img src=\"./images/7.28.png\" alt=\"7.28\" style=\"zoom:50%;\" />\n",
    "\n",
    "non-trivial 的网络服务程序通常会以消息为单位来通信，每条消息有明确的长度与界限。程序每次收到一个完整的消息的时候才开始处理，发送的时候也是把一个完整的消息交给网络库。比如我们前面讲过的 asio chat 服务，它的一条聊天记录就是一条消息。为此我们设计了一个简单的消息格式，即在聊天记录前面加上 4 字节的 length header，`LengthHeaderCodec` 代码及解说见 § 7.3。\n",
    "\n",
    "codec 的基本功能之一是做 TCP 分包：确定每条消息的长度，为消息划分界限。在 non-blocking 网络编程中，codec 几乎是必不可少的。如果只收到了半条消息，那么不会触发消息事件回调，数据会停留在 Buffer 里（数据已经读到 Buffer 中了），等待收到一个完整的消息再通知处理函数。既然这个任务太常见，我们干脆做一个 utility class，避免服务端和客户端程序都要自己处理分包，这就有了 `LengthHeaderCodec`。这个 codec 的使用有点奇怪，不需要继承，它也没有基类，只要把它当成普通 data member 来用，把 `TcpConnection` 的数据“喂”给它，然后向它注册 `onXXXMessage()` 回调，代码见 [asio chat](./codec/codec.h) 示例。muduo 里的 codec 都是这样的风格：通过 `boost::function` 粘合到一起。\n",
    "\n",
    "codec 是一层间接性，它位于 `TcpConnection` 和 `ChatServer` 之间，拦截处理收到的数据显示（`Buffer`），在收到完整的消息之后，解出消息对象（`std::string`），再调用 `CharServer` 对应的处理函数。注意 `CharServer::onStringMessage()` 的参数是 `std::string`，不再是 `muduo::net::Buffer`，也就是说 `LengthHeaderCodec` 把 `Buffer` 解码成了 `string`。另外，在发送消息的时候，`ChatServer` 通过 `LengthHeaderCodec::send()` 来发送 `string`，`LengthHeaderCodec` 负责把它编码成完整的消息。\n",
    "\n",
    "编码成Buffer。这正是“编解码器”名字的由来。消息流程如图7-29所示\n",
    "\n",
    "<img src=\"./images/7.29.png\" alt=\"7.29\" style=\"zoom:50%;\" />\n",
    "\n",
    "Protobuf codec 与此非常类似，只不过消息类型从 `std::string` 变成了 `proto-buf::Message`。对于只接收处理 Query 消息的 `QueryServer` 来说，用 `ProtobufCodec` 非常方便，收到 `protobuf::Message` 之后向下转型成 Query 来用就行（见图 7-30）。\n",
    "\n",
    "<img src=\"./images/7.30.png\" alt=\"7.30\" style=\"zoom:50%;\" />\n",
    "\n",
    "如果要接收处理不止一种消息，ProtobufCodec恐怕还不能单独完成工作，请继续阅读下文。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 实现ProtobufCodec\n",
    "\n",
    "Protobuf 的打包方案我已经在前一节中讲过。编码算法很直截了当，按照前文定义的消息格式一路打包下来，最后更新一下首部的长度即可。代码位于 `examples/protobuf/codec/codec.cc` 中的 [`ProtobufCodec::fillEmptyBuffer()`](./protobuf/codec/codec.cc#fillEmptyBuffer)。\n",
    "\n",
    "解码算法有几个要点：\n",
    "\n",
    "- `protobuf::Message` 是 `new` 出来的对象，它的生命周期如何管理？muduo 采用 `shared_ptr<Message>` 来自动管理对象生命周期，与整体风格保持一致。\n",
    "- 出错如何处理？比如说长度超出范围、check sum 不正确、message type name 不能识别、message parse 出错等等。`ProtobufCodec` 定义了 `ErrorCallback`，用户代码可以注册这个回调。如果不注册，默认的处理是断开连接，让客户重连重试。codec 的单元测试里模拟了各种出错情况。\n",
    "- 如何处理一次收到半条消息、一条消息、一条半消息、两条消息等情况？这是每个 non-blocking 网络程序中的 codec 都要面对的问题。在 p.196-7.3.2 的示例代码中我们已经解决了这个问题。\n",
    "\n",
    "ProtobufCodec 在实际使用中有明显的不足：它只负责把 Buffer 转换为具体类型的 Protobuf Message，每个应用程序拿到 `Message` 对象之后还要再根据其具体类型做一次分发。我们可以考虑做一个简单通用的分发器 dispatcher，以简化客户代码。\n",
    "\n",
    "此外，目前 ProtobufCodec 的实现非常初级，它没有充分利用 `ZeroCopyInputStream` 和 `ZeroCopyOutputStream`，而是把收到的数据作为 byte array 交给 Protobuf Message 去解析，这给性能优化留下了空间。Protobuf Message 不要求数据连续（像 `vector` 那样），只要求数据分段连续（像 `deque` 那样），这给 buffer 管理带来了性能上的好处（避免重新分配内存，减少内存碎片），当然也使得代码变得更为复杂。`muduo::net::Buffer` 非常简单，它内部是 `vector<char>`，我目前不想让 Protobuf 影响 muduo 本身的设计，毕竟 muduo 是通用的网络库，不是为实现 Protobuf RPC 而特制的。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 消息分发器(dispatcher)有什么用\n",
    "\n",
    "前面提到，在使用 TCP 长连接，且在一个连接上传递不止一种 Protobuf 消息的情况下，客户代码需要对收到的消息按类型做分发。比方说，收到 Logon 消息就交给 `QueryServer::onLogon()` 去处理，收到 Query 消息就交给 `QueryServer::onQuery()` 去处理。这个消息分派机制可以做得稍微有点通用性，让所有 muduo+Protobuf 程序受益，而且不增加复杂性。\n",
    "\n",
    "换句话说，又是一层间接性，`ProtobufCodec` 拦截了 `TcpConnection` 的数据，把它转换为 `Message`，`ProtobufDispatcher` 拦截了 `ProtobufCodec` 的 `callback`，按消息具体类型把它分派给多个 `callbacks`，如图 7-31 所示。\n",
    "\n",
    "<img src=\"./images/7.31.png\" alt=\"7.31\" style=\"zoom:50%;\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ProtobufCodec与ProtobufDispatcher的综合运用\n",
    "\n",
    "我写了两个示例代码，client 和 server，把 `ProtobufCodec` 和 `ProtobufDispatcher` 串联起来使用。server 响应 `Query` 消息，发送回 `Answer` 消息，如果收到未知消息类型，则断开连接。client 可以选择发送 `Query` 或 `Empty` 消息，由命令行控制。这样可以测试 `unknown message callback`。\n",
    "\n",
    "为节省篇幅，这里就不列出代码了，见 `[examples/protobuf/codec/{client,server}.cc](./protobuf/codec/codec.h)`。\n",
    "\n",
    "在构造函数中，通过注册回调函数把四方（`TcpConnection`、`codec`、`dispatcher`、`QueryServer`）结合起来。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ProtobufDispatcher的两种实现\n",
    "\n",
    "要完成消息分发，其实就是对消息做 type-switch，这似乎是一个 bad smell，但是 Protobuf Message 的 Descriptor 没有留下定制点（比如暴露一个 `boost::any` 成员），我们只好硬来了。\n",
    "\n",
    "先定义 `ProtobufMessageCallback` 回调：`typedef boost::function<void(Message*)> ProtobufMessageCallback;`\n",
    "\n",
    "注意，本节出现的不是 muduo dispatcher 的真实代码，仅为示意，突出重点，便于画图。\n",
    "\n",
    "`ProtobufDispatcherLite` 的结构非常简单（见图 7-32），它有一个 `map<Descriptor*, ProtobufMessageCallback>` 成员，客户代码可以以 `Descriptor*` 为 key 注册回调（回想：每个具体消息类型都有一个全局的 Descriptor 对象，其地址是不变的，可以用来当 key）。在收到 Protobuf Message 之后，在 map 中找到对应的 `ProtobufMessageCallback`，然后调用之。如果找不到，就调用 `defaultCallback`。\n",
    "\n",
    "<img src=\"./images/7.32.png\" alt=\"7.32\" style=\"zoom:50%;\" />\n",
    "\n",
    "不过，它的设计也有小小的缺陷，那就是`ProtobufMessageCallback`限制了客户代码只能接受基类`Message`，客户代码需要自己做向下转型(down cast)，如图7-33所示。\n",
    "\n",
    "<img src=\"./images/7.33.png\" alt=\"7.33\" style=\"zoom:50%;\" />\n",
    "\n",
    "如果我希望`QueryServer`这么设计：不想每个消息处理函数自己做down cast，而是交给dispatcher去处理，客户代码拿到的就已经是想要的具体类型。接口如图7-34所示。\n",
    "\n",
    "```C++\n",
    "// QueryServer\n",
    "onLogon(Logon*)\n",
    "onQuery(Query*)\n",
    "```\n",
    "\n",
    "那么该如何实现 ProtobufDispatcher 呢？它如何与多个未知的消息类型合作？做 downcast 需要知道目标类型，难道我们要用一长串模板类型参数吗？\n",
    "\n",
    "有一个办法，把多态与模板结合，利用 templated derived class 来提供类型上的灵活性。设计如图 7-35 所示。\n",
    "> 图中画的是 `dynamic_cast`，代码实际上自定义了 `down_cast` 转换操作，在 Debug 编译时会检查动态类型，而在 NDEBUG 编译时会退化为 `static_cast`，没有 RTTI 开销。\n",
    "\n",
    "<img src=\"./images/7.35.png\" alt=\"7.35\" style=\"zoom:50%;\" />\n",
    "\n",
    "ProtobufDispatcher有一个模板成员函数，可以接受注册任意消息类型T的回调，然后它创建一个模板化的派生类`CallbackT<T>`，这样消息的类型信息就保存在了`CallbackT<T>`中，做down cast就简单了。比方说，我们有两个具体消息类型Query和Answer（见图7-36）​。\n",
    "\n",
    "<img src=\"./images/7.36.png\" alt=\"7.36\" style=\"zoom:50%;\" />\n",
    "\n",
    "然后我们这样注册回调：\n",
    "\n",
    "```C++\n",
    "dispatcher_.registerMessageCallback<muduo::Query>(\n",
    "    boost::bind(&QueryServer::onQuery, this, _1, _2, _3));\n",
    "dispatcher_.registerMessageCallback<muduo::Answer>(\n",
    "    boost::bind(&QueryServer::onAnswer, this, _1, _2, _3));\n",
    "```\n",
    "\n",
    "这样会具现化(instantiation)出两个`CallbackT<T>`实体，如图7-37所示。\n",
    "\n",
    "<img src=\"./images/7.37.png\" alt=\"7.37\" style=\"zoom:50%;\" />\n",
    "\n",
    "以上设计参考了`shared_ptr`的`deleter`，[Scott Meyers也谈到过](https://www.artima.com/articles/my-most-important-c-aha-momentsemeverem)。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ProtobufCodec和ProtobufDispatcher有何意义\n",
    "\n",
    "ProtobufCodec和ProtobufDispatcher把每个直接收发Protobuf Message的网络程序都会用到的功能提炼出来做成了公用的utility，这样以后新写Protobuf网络程序就不必为打包分包和消息分发劳神了。它们以库的形式存在，是两个可以拿来就当data member的class。它们没有基类，也没有用到虚函数或者别的什么面向对象特征，不侵犯`muduo::net`或者你的代码。如果不这么做，那将来每个Protobuf网络程序都要自己重新实现类似的功能，徒增负担。\n",
    "\n",
    "§ 9.7 “分布式程序的自动化回归测试”会介绍利用Protobuf的跨语言特性，采用Java为C++服务程序编写test harness。\n",
    "\n",
    "这种编码方案的Java Netty示例代码见http://github.com/chenshuo/muduo-protorpc 中的[com.chenshuo.muduo.codec package](../muduo-protorpc/go/)。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 7.7_限制服务器的最大并发连接数\n",
    "\n",
    "本节以大家都熟悉的`EchoServer`为例，介绍如何限制TCP服务器的并发连接数。代码见[examples/maxconnection/](./maxconnection/)。本节中的“并发连接数”是指一个服务端程序能同时支持的客户端连接数，连接由客户端主动发起，服务端被动接受（`accept(2)`）连接。​（如果要限制应用程序主动发起的连接，则问题要简单得多，毕竟主动权和决定权都在程序本身。​）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 为什么要限制并发连接数\n",
    "\n",
    "一方面，我们不希望服务程序超载；另一方面，更因为 `file descriptor` 是稀缺资源，如果出现 `file descriptor` 耗尽，很棘手，跟 `malloc()` 失败/`new` 抛出 `std::bad_alloc` 差不多同样棘手。\n",
    "\n",
    "我 2010 年 10 月在《分布式系统的工程化开发方法》演讲中曾谈到 libev 的作者 Marc Lehmann 建议的一种应对 “`accept()` 时 `file descriptor` 耗尽” 的办法注。\n",
    "\n",
    "在服务端网络编程中，我们通常用 Reactor 模式来处理并发连接。`listening socket` 是一种特殊的 IO 对象，当有新连接到达时，此 `listening` 文件描述符变得可读 (`POLLIN`)，`epoll_wait` 返回这一事件。然后我们用 `accept(2)` 系统调用获得新连接的 socket 文件描述符。代码主体逻辑如下 (Python)："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "python"
    }
   },
   "outputs": [],
   "source": [
    "serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)\n",
    "serversocket.bind(('', 2007))\n",
    "serversocket.listen(5)\n",
    "serversocket.setblocking(0)\n",
    "\n",
    "poll = select.poll()  # epoll() should work the same\n",
    "poll.register(serversocket.fileno(), select.POLLIN)\n",
    "connections = {}\n",
    "\n",
    "while True:\n",
    "    events = poll.poll(10000)  # 10 seconds L11\n",
    "    for fileno, event in events:\n",
    "        if fileno == serversocket.fileno():\n",
    "            (clientsocket, address) = serversocket.accept() # L14\n",
    "            clientsocket.setblocking(0)\n",
    "            poll.register(clientsocket.fileno(), select.POLLIN)\n",
    "            connections[clientsocket.fileno()] = clientsocket\n",
    "        elif event & select.POLLIN:\n",
    "            # ..."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "假如 L14 的 `accept(2)` 返回 `EMFILE` 该如何应对？这意味着本进程的文件描述符已经达到上限，无法为新连接创建 socket 文件描述符。但是，既然没有 socket 文件描述符来表示这个连接，我们就无法 `close(2)` 它。程序继续运行，回到 L11 再一次调用 `epoll_wait`。这时候 `epoll_wait` 会立刻返回，因为新连接还等待处理，`listening fd` 还是可读的。这样程序立刻就陷入了 busy loop，CPU 占用率接近 100%。这既影响同一 event loop 上的连接，也影响同一机器上的其他服务。\n",
    "\n",
    "该怎么办呢？Marc Lehmann 提到了几种做法：\n",
    "\n",
    "1. 提高进程的文件描述符数目。治标不治本，因为只要有足够多的客户端，就一定能把一个服务进程的文件描述符用完。\n",
    "2. 死等。鸵鸟算法。\n",
    "3. 退出程序。似乎小题大做，为了这种暂时的错误而中断现有的服务似乎不值得。\n",
    "4. 关闭 listening fd。那什么时候重新打开呢？\n",
    "5. 改用 edge trigger。如果漏掉了一次 `accept(2)`，程序再也不会收到新连接。\n",
    "6. 准备一个空闲的文件描述符。遇到这种情况，先关闭这个空闲文件，获得一个文件描述符的名额；再 `accept(2)` 拿到新 socket 连接的描述符；随后立即 `close(2)` 它，这样就优雅地断开了客户端连接；最后重新打开一个空闲文件，把“坑”占住，以备再次出现这种情况时使用。\n",
    "\n",
    "第 2、5 两种做法会导致客户端认为连接已建立，但无法获得服务，因为服务端程序没有拿到连接的文件描述符。\n",
    "\n",
    "muduo 的 Acceptor 正是用第 6 种方案实现的，见 [muduo/net/Acceptor.cc](../muduo/muduo/net/Acceptor.cc)。但是，这个做法在多线程下不能保证正确，会有 race condition。（思考题：是什么 race condition？）\n",
    "\n",
    "其实有另外一种比较简单的办法：file descriptor 是 hard limit，我们可以自己设一个稍低一点的 soft limit，如果超过 soft limit 就主动关闭新连接，这样就可避免触及 “file descriptor 耗尽” 这种边界条件。比方说当前进程的 max file descriptor 是 1024，那么我们可以在连接数达到 1000 的时候进入 “拒绝新连接” 状态，这样就可留给我们足够的腾挪空间。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 在muduo中限制并发连接数\n",
    "\n",
    "在muduo中限制并发连接数的做法简单得出奇。以在§ 6.4.2的EchoServer为例，只需要为它增加一个`int`成员，表示当前的活动连接数。​（如果是多线程程序，应该用`muduo::AtomicInt32`。​）"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "--- ../muduo/examples/simple/echo/echo.h\t2024-08-14 08:59:13.735926400 +0800\n",
      "+++ ../muduo/examples/maxconnection/echo.h\t2024-08-14 08:59:13.716013900 +0800\n",
      "@@ -1,5 +1,5 @@\n",
      "-#ifndef MUDUO_EXAMPLES_SIMPLE_ECHO_ECHO_H\n",
      "-#define MUDUO_EXAMPLES_SIMPLE_ECHO_ECHO_H\n",
      "+#ifndef MUDUO_EXAMPLES_MAXCONNECTION_ECHO_H\n",
      "+#define MUDUO_EXAMPLES_MAXCONNECTION_ECHO_H\n",
      " \n",
      " #include \"muduo/net/TcpServer.h\"\n",
      " \n",
      "@@ -8,9 +8,10 @@\n",
      " {\n",
      "  public:\n",
      "   EchoServer(muduo::net::EventLoop* loop,\n",
      "-             const muduo::net::InetAddress& listenAddr);\n",
      "+             const muduo::net::InetAddress& listenAddr,\n",
      "+             int maxConnections);\n",
      " \n",
      "-  void start();  // calls server_.start();\n",
      "+  void start();\n",
      " \n",
      "  private:\n",
      "   void onConnection(const muduo::net::TcpConnectionPtr& conn);\n",
      "@@ -20,6 +21,8 @@\n",
      "                  muduo::Timestamp time);\n",
      " \n",
      "   muduo::net::TcpServer server_;\n",
      "+  int numConnected_; // should be atomic_int\n",
      "+  const int kMaxConnections_;\n",
      " };\n",
      " \n",
      "-#endif  // MUDUO_EXAMPLES_SIMPLE_ECHO_ECHO_H\n",
      "+#endif  // MUDUO_EXAMPLES_MAXCONNECTION_ECHO_H\n"
     ]
    }
   ],
   "source": [
    "! diff -u ../muduo/examples/simple/echo/echo.h ../muduo/examples/maxconnection/echo.h"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "然后，在`EchoServer::onConnection()`中判断当前活动连接数。如果超过最大允许数，则踢掉连接。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+ #include \"examples/maxconnection/echo.h\"\n",
      "+ using namespace muduo;\n",
      "+ using namespace muduo::net;\n",
      "+ \n",
      "+ EchoServer::EchoServer(EventLoop* loop,\n",
      "+                        const InetAddress& listenAddr,\n",
      "+                        int maxConnections)\n",
      "+   : server_(loop, listenAddr, \"EchoServer\"),\n",
      "+     numConnected_(0),\n",
      "+     kMaxConnections_(maxConnections)\n",
      "+ void EchoServer::onConnection(const TcpConnectionPtr& conn)\n",
      "+ \n",
      "+   if (conn->connected())\n",
      "+   {\n",
      "+     ++numConnected_;\n",
      "+     if (numConnected_ > kMaxConnections_)\n",
      "+     {\n",
      "+       conn->shutdown();\n",
      "+       conn->forceCloseWithDelay(3.0);  // > round trip of the whole Internet.\n",
      "+     }\n",
      "+   }\n",
      "+   else\n",
      "+   {\n",
      "+     --numConnected_;\n",
      "+   }\n",
      "+   LOG_INFO << \"numConnected = \" << numConnected_;\n",
      "+ void EchoServer::onMessage(const TcpConnectionPtr& conn,\n",
      "+                            Buffer* buf,\n",
      "+                            Timestamp time)\n",
      "+   string msg(buf->retrieveAllAsString());\n",
      "+   LOG_INFO << conn->name() << \" echo \" << msg.size() << \" bytes at \" << time.toString();\n"
     ]
    }
   ],
   "source": [
    "! diff --unchanged-line-format= --old-line-format= --new-line-format='+ %L' ../muduo/examples/simple/echo/echo.cc ../muduo/examples/maxconnection/echo.cc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "这种做法可以积极地防止耗尽file descriptor。\n",
    "\n",
    "另外，如果是有业务逻辑的服务，则可以在`shutdown()`之前发送一个简单的响应，表明本服务程序的负载能力已经饱和，提示客户端尝试下一个可用的server（当然，下一个可用的server地址不一定要在这个响应里给出，客户端可以自己去name service查询），这样方便客户端快速failover。\n",
    "\n",
    "§ 7.10将介绍如何处理空闲连接的超时：如果一个连接长时间（若干秒）没有输入数据，则踢掉此连接。办法有很多种，我用timing wheel解决。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# * 7.8 定时器\n",
    "\n",
    "从本节开始的三节内容都与非阻塞网络编程中的定时任务有关"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 程序中的时间\n",
    "\n",
    "程序中对时间的处理是个大问题，在这一节中我先简要谈谈与编程直接相关的内容，把更深入的内容留给日后日期与时间专题文章 注，本书不再细述。\n",
    "\n",
    "在一般的服务器端程序设计中，与时间有关的常见任务有：\n",
    "\n",
    "1. 获取当前时间，计算时间间隔。\n",
    "2. 时区转换与日期计算；把纽约当地时间转换为上海当地时间；2011-02-05之后第100天是几月几号星期几；等等。\n",
    "3. 定时操作，比如在预定的时间执行任务，或者在一段延时之后执行任务。\n",
    "\n",
    "其中第2项看起来比较复杂，但其实最简单。日期计算用[Julian Day Number](https://www.cnblogs.com/Solstice/archive/2010/08/15/1800275.html)，时区转换用[tz database](https://web.cs.ucla.edu/~eggert/tz/tz-link.htm)；唯一麻烦一点的是夏令时，但也可以用tz database解决。这些操作都是纯函数，很容易用一套单元测试来验证代码的正确性。需要特别注意的是，用`tzset/localtime_r`来做时区转换在多线程环境下可能会有问题；对此，我的解决办法是写一个`TimeZone class`，以避免影响全局，日后在日期与时间专题文章中会讲到，本书不再细述。下文不考虑时区，均为UTC时间。\n",
    "\n",
    "真正麻烦的是第1项和第3项。一方面，Linux有一大把令人眼花缭乱的与时间相关的函数和结构体，在程序中该如何选用？另一方面，计算机中的时钟不是理想的计时器，它可能会漂移或跳变。最后，民用的UTC时间与闰秒的关系也让定时任务变得复杂和微妙。当然，与系统当前时间有关的操作也让单元测试变得困难。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Linux时间函数\n",
    "\n",
    "Linux的计时函数，用于获得当前时间：\n",
    "\n",
    "| 接口/结构                                         | 分辨率 | 基准时钟                                  | 主要特性/用途                             | 注意事项                                                     |\n",
    "| ------------------------------------------------- | ------ | ----------------------------------------- | ----------------------------------------- | ------------------------------------------------------------ |\n",
    "| time(2) / time_t                                  | 秒     | CLOCK_REALTIME（墙钟/UTC）                | 最简单、速度快，统计/日志时间戳           | 受NTP/手动校时影响会跳变；32位平台有2038问题                 |\n",
    "| ftime(3) / struct timeb                           | 毫秒   | CLOCK_REALTIME                            | 兼容老代码                                | 已过时，非POSIX推荐，精度有限                                |\n",
    "| gettimeofday(2) / struct timeval                  | 微秒   | CLOCK_REALTIME                            | 常用、开销小                              | 受系统时间调整影响、非单调；POSIX已弃用，建议用 clock_gettime |\n",
    "| clock_gettime(2) / struct timespec                | 纳秒   | 可选：CLOCK_REALTIME / CLOCK_MONOTONIC 等 | 高分辨率；可选单调时钟避免跳变            | 选择 MONOTONIC 可免受校时跳变（频率仍可能微调）；REALTIME 会受校时影响 |\n",
    "| gmtime/localtime/timegm/mktime/strftime/struct tm | N/A    | N/A                                       | 日历时间与字符串相互转换、时区/夏令时处理 | 与“获取当前时间”无直接关系；注意线程安全版本与时区设         |\n",
    "\n",
    "定时函数，用于让程序等待一段时间或安排计划任务：\n",
    "\n",
    "| 接口 | 分辨率 | 基准时钟 | 主要特性/用途 | 注意事项 |\n",
    "|---|---|---|---|---|\n",
    "| sleep(3) | 秒 | 不定（调度器） | 简单等待 | 粗粒度，可能被信号中断而提前返回 |\n",
    "| alarm(2) | 秒 | CLOCK_REALTIME | 若干秒后发送 SIGALRM（一次性） | 进程级、互相覆盖；精度粗，受校时影响 |\n",
    "| usleep(3) | 微秒 | 不定 | 细粒度等待 | 已过时，建议改用 nanosleep |\n",
    "| nanosleep(2) | 纳秒 | 不定（相对睡眠） | 高分辨率相对睡眠；返回剩余时间 | 可被信号打断（EINTR）；受调度影响 |\n",
    "| clock_nanosleep(2) | 纳秒 | 可选：REALTIME/MONOTONIC；支持绝对时间（TIMER_ABSTIME） | 高精度，相对/绝对睡眠 | 用 MONOTONIC 可避免校时跳变；EINTR 处理 |\n",
    "| getitimer(2)/setitimer(2) | 微秒级 | ITIMER_REAL/ITIMER_VIRTUAL/ITIMER_PROF | 传统间隔定时器，触发信号 | 进程级、数量有限；精度一般 |\n",
    "| timer_create/settime/gettime/delete | 纳秒 | 可选：REALTIME/MONOTONIC | POSIX定时器；多定时器；信号或线程方式通知（sigevent） | 接口相对复杂；需要处理信号/回调并发 |\n",
    "| timerfd_create/gettime/settime | 纳秒 | 可选：REALTIME/MONOTONIC | 定时器文件描述符；可与 epoll 集成；read 返回触发次数 | 适合事件循环；注意读出避免丢计数；REALTIME 受校时影响 |\n",
    "\n",
    "我的取舍如下：\n",
    "- 计时 只使用`gettimeofday(2)`来获取当前时间。\n",
    "- 定时 只使用`timerfd_*`系列函数来处理定时任务。\n",
    "\n",
    "`gettimeofday(2)`入选原因（这也是`muduo::Timestamp` class的主要设计考虑）：\n",
    "\n",
    "1. `time(2)`的精度太低，`ftime(3)`已被废弃；`clock_gettime(2)`精度最高，但是其系统调用的开销比`gettimeofday(2)`大。\n",
    "2. 在x86-64平台上，`gettimeofday(2)`不是系统调用，而是在用户态实现的，没有上下文切换和陷入内核的开销 注。\n",
    "3. `gettimeofday(2)`的分辨率（resolution）是1微秒，现在的实现确实能达到这个计时精度，足以满足日常计时的需要。`muduo::Timestamp`用一个`int64_t`来表示从Unix Epoch到现在的微秒数，其范围可达上下30万年。\n",
    "\n",
    "`timerfd_*`入选的原因：\n",
    "1. `sleep(3)`/`alarm(2)`/`usleep(3)`在实现时有可能用了`SIGALRM`信号，在多线程程序中处理信号是个相当麻烦的事情，应当尽量避免，见§4.10。再说，如果主程序和程序库都使用`SIGALRM`，就糟糕了。（为什么？）\n",
    "2. `nanosleep(2)`和`clock_nanosleep(2)`是线程安全的，但是在用阻塞网络编程中，绝对不能用让线程挂起的方式来等待一段时间，这样一来程序会失去响应。正确的做法是注册一个时间回调函数。\n",
    "3. `getitimer(2)`和`timer_create(2)`也是用信号来deliver超时，在多线程程序中也会有麻烦。`timer_create(2)`可以指定信号的接收方是进程还是线程，算是一个进步，不过信号处理函数（signal handler）能做的事情实在很有限。\n",
    "4. `timerfd_create(2)`把时间变成了一个文件描述符，该“文件”在定时器超时的那一刻变得可读，这样就能很方便地融入`select(2)`/`poll(2)`框架中，用统一的方式来处理I/O事件和超时事件，这也是Reactor模式的长处。我在以前发表的《Linux新增系统调用的启示》注 中也谈到了这个想法，现在把这个想法在muduo网络库中实现了。\n",
    "5. 传统的Reactor利用`select(2)`/`poll(2)`/`epoll(4)`的timeout来实现定时功能，但`poll(2)`和`epoll_wait(2)`的定时精度只有毫秒，远低于`timerfd_settime(2)`的定时精度。\n",
    "\n",
    "必须要说明，在Linux这种相对实时多任务操作系统中，在用户态实现完全精确可控的计时和定时是做不到的，因为当前任务可能会被随时切换出去，这在CPU负载大的时候尤为明显。但是，我们的程序可以尽量提高时间精度，必要的时候通过控制CPU负载来提高时间操作的可靠性，让程序在99.99%的时候都是按预期执行的。这或许比把用实时操作系统并重新写用程序及测试代码要经济一些。\n",
    "\n",
    "关于时间的精度（accuracy）问题我曾经留到日期与时间专题文章中讨论，本书不再细述，它与分辨率（resolution）不完全是一回事儿。时间跳变和调秒的影响与应对也不在此展开讨论了。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## muduo的定时器接口\n",
    "\n",
    "muduo EventLoop有三个定时器函数：\n",
    "- [EventLoop.h](../muduo/muduo/net/EventLoop.cc#runAt)\n",
    "\n",
    "函数名称很好地反映了其用途：\n",
    "\n",
    "- `runAt` 在指定的时间调用 `TimerCallback`；\n",
    "- `runAfter` 等一段时间调用 `TimerCallback`；\n",
    "- `runEvery` 以固定的间隔反复调用 `TimerCallback`；\n",
    "- `cancel` 取消 `timer`。\n",
    "\n",
    "回调函数在 `EventLoop` 对象所属的线程发生，与 `onMessage()`、`onConnection()` 等网络事件函数在同一个线程。muduo 的 `TimerQueue` 采用了平衡二叉树来管理未到期的 `timers`，因此这些操作的事件复杂度是 `O(log N)`。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Boost.Asio Timer示例\n",
    "\n",
    "[Boost.Asio教程](https://www.boost.org/doc/libs/latest/doc/html/boost_asio/tutorial.html) 里以`Timer`和`Daytime`为例介绍Asio的基本使用，`daytime`已经在§7.1中介绍过，这里着重谈谈`Timer`。Asio有5个`Timer`示例，muduo把其中四个重新实现了一遍，并扩充了第5个示例。\n",
    "\n",
    "1. 阻塞式的定时，muduo不支持这种用法，无代码。\n",
    "2. 非阻塞定时，见`examples/asio/tutorial/timer2`。\n",
    "3. 在TimerCallback里传递参数，见`examples/asio/tutorial/timer3`。\n",
    "4. 以成员函数为TimerCallback，见`examples/asio/tutorial/timer4`。\n",
    "5. 在多线程中回调，用mutex保护共享变量，见`examples/asio/tutorial/timer5`。\n",
    "6. 在多线程中回调，缩小临界区，把不需要互斥执行的代码移出来，见`examples/asio/tutorial/timer6`。\n",
    "\n",
    "为节省篇幅，这里只列出`timer4`。这个程序的功能是以1秒为间隔打印5个整数，乍看来代码有点小题大做，但是值得注意的是定时器事件与IO事件是在同一线程发生的，程序就像处理IO事件一样处理超时事件。\n",
    "\n",
    "[timer.cc](../muduo/muduo/examples/asio/tutorial/timer4/timer.cc)\n",
    "\n",
    "最后再强调一遍，在非阻塞服务端编程中，绝对不能用`sleep()`或类似的办法来让程序原地停留等待，这会让程序失去响应，因为主事件循环被挂起了，无法处理IO事件。这就像在Windows编程中绝对不能在消息循环里执行耗时的代码是一个道理，这会让程序界面失去响应。Reactor模式的网络编程确有些类似传统的消息驱动的Windows编程。对于“定时”任务，把它变成一个特定的消息，到时候触发相应的消息处理函数就行了。\n",
    "\n",
    "> AI:\n",
    "> - 如果你有 **专门的工作线程**（不跑事件循环，而是处理某些独立任务），那在这些线程里使用 `sleep_for` 是没问题的。它只会阻塞这个工作线程，不会影响事件循环。\n",
    "> - 但只要是 **负责事件分发/回调** 的线程，就必须保持非阻塞。想要“等待某个时间点”，应该通过事件循环提供的 **定时器机制** 或 **异步回调** 来实现，让事件循环帮你在将来某个时刻重新调度处理逻辑\n",
    "\n",
    "Boost.Asio的timer示例只用到了`EventLoop::runAfter`，我再举一个`EventLoop::runEvery`的例子。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Java Netty示例\n",
    "\n",
    "Netty是一个非常好的Java NIO网络库，它附带的示例程序有echo和discard两个简单网络协议。与§ 7.1不同，Netty版的echo和discard服务端有流量统计功能，这需要用到固定间隔的定时器(`EventLoop::runEvery`)。\n",
    "\n",
    "其client的代码类似前文的chargen，为节省篇幅，请阅读源码[examples/netty/discard/client.cc](./netty/discard/client.cc)。\n",
    "\n",
    "这里列出discard server的完整代码。代码整体结构上与§ 6.4.2的EchoServer差别不大，这算是简单网络服务器的典型模式了。\n",
    "\n",
    "DiscardServer可以配置成多线程服务器，muduo `TcpServer`有一个内置的one loop per thread多线程IO模型，可以通过`setThreadNum()`来开启。\n",
    "\n",
    "构造函数注册了一个间隔为3秒的定时器，调用`DiscardServer::printThroughput()`打印出吞吐量。消息回调只比p.178的代码多两行，用于统计收到的数据长度和消息次数。\n",
    "- [server.cc](./netty/discard/server.cc#L31)\n",
    "\n",
    "消息回调只比p.178的代码多两行，用于统计收到的数据长度和消息次数。\n",
    "- [server.cc](./netty/discard/server.cc#onMessage)\n",
    "\n",
    "在每一个统计周期，打印数据吞吐量。\n",
    "- [server.cc](./netty/discard/server.cc#printThroughput)\n",
    "\n",
    "以下是数据成员，注意用了整数的原子操作`AtomicInt64`来记录收到的字节数和消息数，这是为了多线程安全性。\n",
    "- [server.cc](./netty/discard/server.cc#L75)\n",
    "\n",
    "`main()`函数，有一个可选的命令行参数，用于指定线程数目。\n",
    "- [server.cc](./netty/discard/server.cc#main)\n",
    "\n",
    "运行方法，在同一台机器的两个命令行窗口分别运行："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "$ ./netty_discard_server \n",
    "20251013 16:59:59.288402Z 69360 INFO  pid = 69360, tid = 69360 - server.cc:83\n",
    "20251013 16:59:59.289295Z 69360 INFO  starting 0 threads. - server.cc:37\n",
    "0.000 MiB/s 0.000 Ki Msgs/s   -nan bytes per msg\n",
    "0.000 MiB/s 0.000 Ki Msgs/s   -nan bytes per msg\n",
    "0.000 MiB/s 0.000 Ki Msgs/s   -nan bytes per msg\n",
    "0.000 MiB/s 0.000 Ki Msgs/s   -nan bytes per msg\n",
    "0.000 MiB/s 0.000 Ki Msgs/s   -nan bytes per msg\n",
    "0.000 MiB/s 0.000 Ki Msgs/s   -nan bytes per msg\n",
    "0.000 MiB/s 0.000 Ki Msgs/s   -nan bytes per msg\n",
    "20251013 17:00:22.782650Z 69360 INFO  TcpServer::newConnection [DiscardServer] - new connection [DiscardServer-0.0.0.0:2009#1] from 127.0.0.1:44482 - TcpServer.cc:80\n",
    "0.763 MiB/s 3.002 Ki Msgs/s 260.28 bytes per msg\n",
    "4.869 MiB/s 18.978 Ki Msgs/s 262.70 bytes per msg\n",
    "4.813 MiB/s 18.858 Ki Msgs/s 261.32 bytes per msg\n",
    "4.902 MiB/s 19.216 Ki Msgs/s 261.24 bytes per msg\n",
    "5.162 MiB/s 20.236 Ki Msgs/s 261.18 bytes per msg\n",
    "20251013 17:00:36.332389Z 69360 INFO  TcpServer::removeConnectionInLoop [DiscardServer] - connection DiscardServer-0.0.0.0:2009#1 - TcpServer.cc:109\n",
    "1.757 MiB/s 6.893 Ki Msgs/s 261.02 bytes per msg\n",
    "0.000 MiB/s 0.000 Ki Msgs/s   -nan bytes per msg\n",
    "0.000 MiB/s 0.000 Ki Msgs/s   -nan bytes per msg\n",
    "0.000 MiB/s 0.000 Ki Msgs/s   -nan bytes per msg"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "$ ./netty_discard_client 127.0.0.1 256"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "改变第二个命令的最后一个参数（上面的`256`）​，可以观察不同的消息大小对吞吐量的影响。\n",
    "\n",
    "- 练习1：把二者的关系绘制成函数曲线，看看有什么规律，想想为什么。\n",
    "- 练习2：在局域网的两台机器上运行客户端和服务端，找出让吞吐量达到最大的消息长度。这个数字与练习1中的相比是大还是小？为什么？\n",
    "\n",
    "有兴趣的读者可以对比一下Netty的吞吐量，muduo应该能轻松取胜。\n",
    "\n",
    "discard client/server测试的是单向吞吐量，echo client/server测试的是双向吞吐量。\n",
    "\n",
    "这两个服务端都支持多个并发连接，两个客户端都是单连接的。前文§ 6.5实现了一个pingpong协议，客户端和服务端都是多连接，用来测试muduo在多线程大量连接情况下的性能表现。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 测量两台机器的网络延迟和时间差\n",
    "\n",
    "本节介绍一个简单的网络程序roundtrip，用于测量两台机器之间的网络延迟，即“往返时间(round trip time，RTT)”。其主要考察定长TCP消息的分包与`TCP_NODELAY`的作用。本节的代码见[examples/roundtrip/roundtrip.cc](./roundtrip/roundtrip.cc)。\n",
    "\n",
    "测量round trip time的办法很简单：\n",
    "- host A发一条消息给host B，其中包含host A发送消息的本地时间。\n",
    "- host B收到之后立刻把消息echo回host A。\n",
    "- host A收到消息之后，用当前时间减去消息中的时间就得到了round trip time。\n",
    "\n",
    "NTP协议的工作原理与之类似注，不过，除了测量round trip time，NTP还需要知道两台机器之间的时间差(clock offset)，这样才能校准时间。\n",
    "\n",
    "图7-38是NTP协议收发消息的协议，$round trip time=(T_4−T_1)−(T_3−T_2)$，NTP的要求是往返路径上的单程延迟要尽量相等，这样才能减少系统误差。偶然误差由单程延迟的不确定性决定。\n",
    "\n",
    "<img src=\"./images/7.38.png\" alt=\"7.38\" style=\"zoom:50%;\" />\n",
    "\n",
    "在我设计的roundtrip示例程序中，协议有所简化，如图7-39所示。\n",
    "\n",
    "<img src=\"./images/7.39.png\" alt=\"7.39\" style=\"zoom:50%;\" />\n",
    "\n",
    "计算公式如下。\n",
    "- $\\text{round trip time} = T_3 - T_1$\n",
    "- $\\text{clock offset} = T_2 - \\frac{T_1 + T_3}{2}$\n",
    "\n",
    "简化之后的协议少取一次时间，因为server收到消息之后立刻发送回client，耗时很少（若干微秒）​，基本不影响最终结果。我设计的消息格式是16字节定长消息，如图7-40所示。\n",
    "\n",
    "<img src=\"./images/7.40.png\" alt=\"7.40\" style=\"zoom:50%;\" />\n",
    "\n",
    "$T_1$和$T_2$都是`muduo::Timestamp`，成员是一个`int64_t`，表示从Unix Epoch到现在的微秒数。为了让消息的单程往返时间接近，server和client发送的消息都是16bytes，这样做到对称。由于是定长消息，可以不必使用codec，在message callback中直接用\n",
    "\n",
    "```cpp\n",
    "while (buffer->readableBytes() >= frameLen) {\n",
    "  // ...\n",
    "}\n",
    "\n",
    "就能decode。请读者思考：如果把`while`换成`if`会有什么后果？\n",
    "\n",
    "client程序以200ms为间隔发送消息，在收到消息之后打印round trip time和clock offset。一次运作实例如图7-41所示。\n",
    "\n",
    "<img src=\"./images/7.41.png\" alt=\"7.41\" style=\"zoom:50%;\" />\n",
    "\n",
    "在这个例子中，client和server各自的本地时钟不是完全对准的，server的时间快了850µs，用roundtrip程序能测量出这个时间差。有了这个时间差，就能校正分布式系统中测量得到的消息延迟。\n",
    "\n",
    "比方说以图7-41为例，server在它本地1.235000s时刻发送了一条消息，client在它本地1.234300s收到这条消息，若直接计算的话还是-700μs。这个结果肯定是错的，因为server和client不在一个时钟域（clock domain，这是数字电路中的概念），它们的时间直接相减无意义。如果我们已经测量得到server比client快850μs，那么用这个数据做一次校正：-700+850=150μs，这个结果就比较符合实际了。当然，在实际应用中，clock offset要经过一个低通滤波才能使用，不然偶然性太大。\n",
    "\n",
    "请读者思考：为什么不能直接以RTT/2作为两台机器之间收发消息的单程延迟？这个数字是偏大还是偏小？\n",
    "\n",
    "这个程序在局域网中使用没有问题；如果在广域网上使用，而且RTT大于200ms，那么受Nagle算法影响，测量结果是错误的。因为应用程序记录的发包时间与操作系统真正发出数据包的时间之差不再是一个可以忽略的小间隔。具体分析留作练习，这能测试读者对Nagle的理解。这时候我们需要设置TCP_NODELAY参数，让程序在广域网上也能正常工作。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 7.10 用timing wheel踢掉空闲连接\n",
    "\n",
    "本节介绍如何使用timing wheel来踢掉空闲的连接。一个连接如果若干秒没有收到数据，就被认为是空闲连接。本文的代码见[examples/idleconnection](./idleconnection/)。\n",
    "\n",
    "在严肃的网络程序中，应用层的心跳协议是必不可少的。应该用心跳消息来判断对方进程是否能正常工作，“踢掉空闲连接”只是一时的权宜之计。我这里想顺便讲`shared_ptr`和`weak_ptr`的用法。\n",
    "\n",
    "如果一个连接连续几秒（后文以8s为例）内没有收到数据，就把它断开，为此有两种简单、粗暴的做法：\n",
    "- 每个连接保存“最后收到数据的时间`lastReceiveTime`”，然后用一个定时器，每秒遍历一遍所有连接，断开那些`(now-connection.lastReceiveTime) > 8s`的connection。这种做法全局只有一个repeated timer，不过每次`timeout`都要检查全部连接，如果连接数目比较大（几十上万），这一步可能会比较费时。\n",
    "- 每个连接设置一个`one-shot timer`，超时定为8s，在超时的时候就断开本连接。当然，每次收到数据要去更新timer。这种做法需要很多`one-shot timer`，会频繁地更新`timers`。如果连接数目比较大，可能对EventLoop的`TimerQueue`造成压力。\n",
    "\n",
    "使用timing wheel能避免上述两种做法的缺点。timing wheel可以翻译为“时间轮盘”或“刻度盘”，本文保留英文。\n",
    "\n",
    "连接超时不需要精确定时，只要大致8秒超时断开就行，多一秒、少一秒关系不大。处理连接超时可用一个简单的数据结构：8个桶组成的循环队列。第1个桶放1秒之后将要超时的连接，第2个桶放2秒之后将要超时的连接。每个连接一收到数据就把自己放到第8个桶，然后在每秒的timer里把第一个桶里的连接断开，把这个空桶挪到队尾。这样大致可以做到8秒没有数据就超时断开连接。更重要的是，每次不再用检查全部的连接，只要检查第一个桶里的连接，相当于把任务分散了。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## timing wheel原理\n",
    "\n",
    "《Hashed and hierarchical timing wheels: efficient data structures for implementing a timer facility》 这篇论文详细比较了实现定时器的各种数据结构，并提出了层次化的timing wheel与hash timing wheel等新结构。针对本节要解决的问题的特点，我们不需要实现一个通用的定时器，只用实现simple timing wheel即可。\n",
    "\n",
    "simple timing wheel的基本结构是一个循环队列，还有一个指向队尾的指针(tail)，这个指针每秒移动一格，就像钟表上的时针，timing wheel由此得名。\n",
    "\n",
    "以下是某一时刻timing wheel的状态（见图7-42的左图），格子里的数字是倒计时（与通常的timing wheel相反），表示这个格子（桶子）中连接的剩余寿命。\n",
    "\n",
    "<img src=\"./images/7.42.png\" alt=\"7.42\" style=\"zoom:50%;\" />\n",
    "\n",
    "1秒以后（见图7-42的右图）​，tail指针移动一格，原来四点钟方向的格子被清空，其中的连接已被断开。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 连接超时被踢掉的过程\n",
    "\n",
    "假设在某个时刻，`conn 1`到达，把它放到当前格子中，它的剩余寿命是7秒（见图7-43的左图）​。此后`conn 1`上没有收到数据。1秒之后（见图7-43的右图）​，`tail`指向下一个格子，`conn 1`的剩余寿命是6秒。\n",
    "\n",
    "<img src=\"./images/7.43.png\" alt=\"7.43\" style=\"zoom:50%;\" />\n",
    "\n",
    "又过了几秒，`tail`指向`conn 1`之前的那个格子，`conn 1`即将被断开（见图7-44的左图）​。下一秒（见图7-44的右图）​，`tail`重新指向`conn 1`原来所在的格子，清空其中的数据，断开`conn 1`连接。\n",
    "\n",
    "<img src=\"./images/7.44.png\" alt=\"7.44\" style=\"zoom:50%;\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 连接刷新\n",
    "\n",
    "如果在断开`conn 1`之前收到数据，就把它移到当前的格子里。`conn 1`的剩余寿命是3秒（见图7-45的左图）​，此时`conn 1`收到数据，它的寿命恢复为7秒（见图7-45的右图）​。\n",
    "\n",
    "<img src=\"./images/7.45.png\" alt=\"7.45\" style=\"zoom:50%;\" />\n",
    "\n",
    "时间继续前进，`conn 1`寿命递减，不过它已经比第一种情况长寿了（见图7-46）​。\n",
    "\n",
    "<img src=\"./images/7.46.png\" alt=\"7.46\" style=\"zoom:50%;\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 多个连接\n",
    "\n",
    "timing wheel中的每个格子是个`hash set`，可以容纳不止一个连接。\n",
    "\n",
    "比如一开始，`conn 1`到达。随后，`conn 2`到达（见图7-47）​，这时候`tail`还没有移动，两个连接位于同一个格子中，具有相同的剩余寿命。​（在图7-47中画成链表，代码中是哈希表。​）\n",
    "\n",
    "<img src=\"./images/7.47.png\" alt=\"7.47\" style=\"zoom:50%;\" />\n",
    "\n",
    "几秒之后，`conn 1`收到数据，而`conn 2`一直没有收到数据，那么`conn 1`被移到当前的格子中。这时`conn 1`的预期寿命比`conn 2`长（见图7-48）​。\n",
    "\n",
    "<img src=\"./images/7.48.png\" alt=\"7.48\" style=\"zoom:50%;\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 代码实现与改进\n",
    "\n",
    "我们用以前多次出现的EchoServer来说明具体如何实现timing wheel。代码见[examples/idleconnection](./idleconnection/)。\n",
    "\n",
    "在具体实现中，格子里放的不是连接，而是一个特制的`Entry struct`，每个`Entry`包含`TcpConnection`的`weak_ptr`。`Entry`的析构函数会判断连接是否还存在（用`weak_ptr`）​，如果还存在则断开连接。\n",
    "\n",
    "数据结构：​（本节的代码压缩了单行缩进）\n",
    "- [echo.h](./idleconnection/echo.h#34)\n",
    "\n",
    "在实现中，为了简单起见，我们不会真的把一个连接从一个格子移到另一个格子，而是采用引用计数的办法，用`shared_ptr`来管理`Entry`。如果从连接收到数据，就把对应的`EntryPtr`放到这个格子里，这样它的引用计数就递增了。当`Entry`的引用计数递减到零时，说明它没有在任何一个格子里出现，那么连接超时，`Entry`的析构函数会断开连接。\n",
    "\n",
    "注意在头文件中我们自己定义了`shared_ptr<T>`的`hash`函数，原因是直到Boost 1.47.0之前，`unordered_set<shared_ptr<T>>`虽然可以编译通过，但是其`hash_value`是`shared_ptr`隐式转换为`bool`的结果。也就是说，如果不自定义`hash`函数，那么`unordered_{set/map}`会退化为链表。\n",
    "\n",
    "timing wheel用`boost::circular_buffer`实现，其中每个`Bucket`元素是个hash set of `EntryPtr`。\n",
    "\n",
    "在构造函数中，注册每秒的回调（`EventLoop::runEvery()`注册`EchoServer::onTimer()`），然后把timing wheel设为适当的大小。\n",
    "- [echo.cc](./idleconnection/echo.cc#13)\n",
    "\n",
    "其中，`EchoServer::onTimer()`的实现只有一行：往队尾添加一个空的Bucket，这样`circular_buffer`会自动弹出队首的`Bucket`，并析构之。在析构`Bucket`的时候，会依次析构其中的`EntryPtr`对象，这样`Entry`的引用计数就不用我们去操心，C++的值语意会帮我们搞定一切。\n",
    "- [echo.cc](./idleconnection/echo.cc#74)\n",
    "\n",
    "在连接建立时，创建一个`Entry`对象，把它放到timing wheel的队尾。另外，我们还需要把`Entry`的弱引用保存到`TcpConnection`的`context`里，因为在收到数据的时候还要用到`Entry`。​（思考题：如果`TcpConnection::setContext`保存的是强引用`EntryPtr`，会出现什么情况？​）\n",
    "- [echo.cc](./idleconnection/echo.cc#33)\n",
    "\n",
    "在收到消息时，从`TcpConnection`的`context`中取出`Entry`的弱引用，把它提升为强引用`EntryPtr`，然后放到当前的timing wheel队尾。​（思考题：为什么要把`Entry`作为`TcpConnection`的`context`保存，如果这里再创建一个新的`Entry`会有什么后果？​）\n",
    "- [echo.cc](./idleconnection/echo.cc#55)\n",
    "\n",
    "然后呢？没有然后了，程序已经完成了我们想要的功能。​（完整的代码会调用`dumpConnectionBuckets()`来打印`circular_buffer`变化的情况，运行一下即可理解。​）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 改进\n",
    "\n",
    "在现在的实现中，每次收到消息都会往队尾添加`EntryPtr`（当然，hash set会帮我们去重(deduplication)）。一个简单的改进措施是，在`TcpConnection`里保存“最后一次往队尾添加引用时的`tail`位置”，收到消息时先检查`tail`是否变化，若无变化则不重复添加`EntryPtr`，若有变化则把`EntryPtr`从旧的`Bucket`移到当前队尾`Bucket`。这样或许能提高空间和时间效率。以上改进留作练习。\n",
    "\n",
    "另外一个思路是“选择排序”：使用链表将`TcpConnection`串起来，`TcpConnection`每次收到消息就把自己移到链表末尾，这样链表是按接收时间先后排序的。再用一个定时器定期从链表前端查找并踢掉超时的连接。代码示例位于同一目录。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 7.11 简单的消息广播服务"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "本节介绍用muduo实现一个简单的topic-based消息广播服务，这其实是“聊天室”的一个简单扩展，不过聊的不是人，而是分布式系统中的程序。本节的代码见`examples/hub`。\n",
    "\n",
    "在分布式系统中，除了常用的end-to-end通信，还有一对多的广播通信。一提到“广播”，或许会让人联想到IP多播或IP组播，这不是本节的主题。本节将要谈的是基于TCP协议的应用层广播。示意图如图7-49所示。\n",
    "\n",
    "<img src=\"./images/7.49.png\" alt=\"7.49\" style=\"zoom:50%;\" />\n",
    "\n",
    "图7-49中的圆角矩形代表程序，“Hub”是一个服务程序，不是网络集线器，它起到类似集线器的作用，故而得名。Publisher和Subscriber通过TCP协议与Hub程序通信。\n",
    "\n",
    "`Publisher`把消息发送到某个topic上，Subscriber订阅该topic，然后就能收到消息。即Publisher借助Hub把消息广播给了一个或多个Subscriber。这种pub/sub结构的好处在于可以增加多个Subscriber而不用修改Publisher，一定程度上实现了“解耦”（也可以看成分布式的Observer pattern）。由于走的是TCP协议，广播是基本可靠的，这里的“可靠”指的是“比UDP可靠”，不是“完全可靠”。（思考：如何避免Hub成为single point of failure?）\n",
    "\n",
    "应用层广播在分布式系统中用处很大，这里略举几例。\n",
    "- 体育比分转播: 有8片比赛场地正在进行羽毛球比赛，每个场地的计分程序把当前比分发送到各自的`topic`上（第1号场地发送到`court1`，第2号场地发送到`court2`，依此类推）。需要用到比分的程序（赛场的大屏幕显示、网上比分转播等）自己订阅感兴趣的`topic`，就能及时收到最新比分数据。由于本节实现的不是100%可靠广播，那么消息应该是`snapshot`，而不是`delta`。（换句话说，消息的内容是“现在是几比几”，而不是“刚才谁得分”。）\n",
    "- 负载监控: 每台机器上运行一个监控程序，周期性地把本机当前负载（CPU、网络、磁盘、温度）`publish`到以`hostname`命名的`topic`上，这样需要用到这些数据的程序只要在`Hub`订阅相应的`topic`就能获得数据，无须与多台机器直接打交道。（为了可靠起见，监控程序发送的消息中应该包含时间戳，这样能防止过期(`stale`)数据，甚至一定程度上起到心跳的作用。）沿着这个思路，分布式系统中的服务程序也可以把自己的当前负载发布到`Hub`上，供`load balancer`和`monitor`取用。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 协议\n",
    "\n",
    "为了简单起见，muduo的Hub示例采用以“`\\r\\n`”分界的文本协议，这样用telnet就能测试Hub。协议只有以下三个命令：\n",
    "\n",
    "\n",
    "- `sub<topic>\\r\\n`: 该命令表示订阅`<topic>`，以后该topic有任何更新都会发给这个TCP连接。在sub的时候，Hub会把该`<topic>`上最近的消息发给此`Subscriber`。\n",
    "- `unsub<topic>\\r\\n`: 该命令表示退订`<topic>`。\n",
    "- `pub<topic>\\r\\n<content>\\r\\n`: 往`<topic>`发送消息，内容为`<content>`。所有订阅了此`<topic>`的Subscriber会收到同样的消息，“`pub<topic>\\r\\n<content>\\r\\n`”。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 代码\n",
    "\n",
    "muduo示例中的Hub分为几个部分：\n",
    "- Hub服务程序，负责一对多的消息分发。它会记住每个client订阅了哪些topic，只把消息发给特定的订阅者。代码参见[examples/hub/hub.cc](./hub/hub.cc)。\n",
    "- pubsub库，为了方便编写使用Hub服务的应用程序，我写了一个简单的client library，用来和Hub打交道。这个library可以订阅topic、退订topic、往指定的topic发布消息。代码参见[examples/hub/pubsub.{h， cc}](./hub/pubsub.h)。\n",
    "- sub示例程序，这个命令行程序订阅一个或多个topic，然后等待Hub的数据。代码参见[examples/hub/sub.cc](./hub/sub.cc)。\n",
    "- pub示例程序，这个命令行程序往某个topic发布一条消息，消息内容由命令行参数指定。代码参见[examples/hub/pub.cc](./hub/pub.cc)。\n",
    "\n",
    "一个程序可以既是Publisher又是Subscriber，而且pubsub库只用一个TCP连接（这样failover比较简便）。使用范例如下所示。\n",
    "\n",
    "1. 开启4个命令行窗口。\n",
    "2. 在第一个窗口运行\n",
    "```bash\n",
    "$ hub 9999\n",
    "```\n",
    "3. 在第二个窗口运行\n",
    "```bash\n",
    "$ sub 127.0.0.1：9999 mytopic\n",
    "```\n",
    "4. 在第三个窗口运行\n",
    "```bash\n",
    "$ sub 127.0.0.1：9999 mytopic court\n",
    "```\n",
    "5. 在第四个窗口运行\n",
    "```bash\n",
    "$ pub 127.0.0.1：9999 mytopic \"Hello world.\"\n",
    "```\n",
    "这时第二、三号窗口都会打印“mytopic:  Hello world.”，表明收到了mytopic这个主题上的消息。\n",
    "6. 在第四个窗口运行\n",
    "```bash\n",
    "$ pub 127.0.0.1：9999 court \"13：11\"\n",
    "```\n",
    "这时第三号窗口会打印“court：13：11”，表明收到了court这个主题上的消息。第二号窗口没有订阅此消息，故无输出。\n",
    "\n",
    "借助这个简单的pub/sub机制，还可以做很多有意思的事情。比如把分布式系统中的程序的一部分end-to-end通信改为通过pub/sub来做（例如，原来是A向B发一个SOAP request，B通过同一个TCP连接发回response（分析二者的通信只能通过查看log或用tcpdump截获）；现在是A在`topic_a_to_b`上发布request，B在`topic_b_to_a`上发response），这样多挂一个`monitoring subscriber`就能轻易地查看通信双方的沟通情况，很容易做状态监控与trouble shooting。\n",
    "> Troubles hooting 的意思是：故障排查、问题诊断与定位，并采取措施解决问题。常用于 IT、网络、设备维护等场景。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 多线程的高效广播\n",
    "\n",
    "在本节这个例子中，Hub是个单线程程序。假如有一条消息要广播给1000个订阅者，那么只能一个一个地发，第1个订阅者收到消息和第1000个订阅者收到消息的时差可以长达若干毫秒。那么，有没有办法提高速度、降低延迟呢？我们当然会想到用多线程。但是简单的办法并不一定能奏效，因为一个全局锁就把多线程程序退化为单线程执行。为了真正提速，我想到了用thread local的办法，比如把1000个订阅者分给4个线程，每个线程的操作基本都是无锁的，这样可以做到并行地发送消息。示例代码见[examples/asio/chat/server_threaded_highperformance.cc](./asio/chat/server_threaded_highperformance.cc)。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 7.12 “串并转换”连接服务器及其自动化测试\n",
    "\n",
    "本节介绍如何使用test harness来测试一个具有内部逻辑的网络服务程序。这是一个既扮演服务端，又扮演客户端的网络程序。代码见[examples/multiplexer](./multiplexer/)。\n",
    "\n",
    "云风在他的博客中提到了网游连接服务器的功能需求，我用C++初步实现了这些需求，并为之编写了配套的自动化test harness，作为muduo网络库的示例。\n",
    "> 在 http://blog.codingnow.com/2010/11/go_prime.html 搜“练手项目”。\n",
    "\n",
    "注意：本节呈现的代码仅仅实现了基本的功能需求，没有考虑安全性，也没有特别优化性能，不适合用作真正的放在公网上运行的网游连接服务器。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 功能需求\n",
    "\n",
    "这个连接服务器把多个客户连接汇聚为一个内部TCP连接，起到“数据串并转换”的作用，让backend的逻辑服务器专心处理业务，而无须顾及多连接的并发性。系统的框图如图7-50所示。\n",
    "\n",
    "<img src=\"./images/7.50.png\" alt=\"7.50\" style=\"zoom:50%;\" />\n",
    "\n",
    "这个连接服务器的作用与数字电路中的数据选择器(`multiplexer`)类似（见图7-51）​，所以我把它命名为`multiplexer`。​（其实IO multiplexing也是取的这个意思，让一个thread-of-control能有选择地处理多个IO文件描述符。​）\n",
    "\n",
    "<img src=\"./images/7.51.png\" alt=\"7.51\" style=\"zoom:50%;\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 实现\n",
    "\n",
    "multiplexer的功能需求不复杂，无非是在backend connection和client connections之间倒腾数据。对每个新client connection分配一个新的整数id，如果id用完了，则断开新连接（这样通过控制id的数目就能控制最大连接数）。另外，为了避免id过快地复用（有可能造成backend串话），multiplexer采用queue来管理free id，每次从队列的头部取id，用完之后放回queue的尾部。具体来说，主要是处理四种事件：\n",
    "\n",
    "- 当client connection到达或断开时，向backend发出通知。代码见[`onClientConnection()`](./multiplexer/multiplexer_simple.cc#L50)。\n",
    "- 当从client connection收到数据时，把数据连同connection id一同发给backend。代码见[`OnClientMessage()`](./multiplexer/multiplexer_simple.cc#L113)。\n",
    "- 当从backend connection收到数据时，辨别数据是发给哪个client connection，并执行相应的转发操作。代码见[`OnBackendMessage()`](./multiplexer/multiplexer_simple.cc#L191)。\n",
    "- 如果backend connection断开连接，则断开所有client connections（假设client会自动重试）。代码见[`OnBackendConnection()`](./multiplexer/multiplexer_simple.cc#L169)。\n",
    "\n",
    "由上可见，multiplexer的功能与proxy颇为类似。[`multiplexer_simple.cc`](./multiplexer/multiplexer_simple.cc)是一个单线程版的实现，借助muduo的multiplexing特性，可以方便地处理多个并发连接。多线程版的实现见[`multiplexer.cc`](./multiplexer/multiplexer.cc)。\n",
    "\n",
    "在实现的时候有以下两点值得注意。\n",
    "\n",
    "TcpConnection的id如何存放？当从backend收到数据，如何根据id找到对应的client connection？当从client connection收到数据，如何得知其id？\n",
    "\n",
    "- 第一个问题比较好解决，用：`std::map<int, TcpConnectionPtr> clientConns`保存从id到client connection的映射就行。\n",
    "- 第二个问题固然可以用类似的办法解决，但是我想借此介绍一下`muduo::net::TcpConnection`的context功能。每个TcpConnection都有一个`boost::any`成员，可由客户代码自由支配(get/set)，代码如下。这个`boost::any`是TcpConnection的context，可以用于保存与connection绑定的任意数据（比方说connection id，connection的最后数据到达时间，connection所代表的用户的名字等等）。这样客户代码不必继承TcpConnection就能attach自己的状态，而且也用不着TcpConnectionFactory了（如果不允许继承，那么必须要向TcpServer注入此factory）。\n",
    "\n",
    "- [TcpConnection.h](../muduo/net/TcpConnection.h#L41)\n",
    "\n",
    "对于multiplexer，在`onClientConnection()`里调用`conn->setContext(id)`，把id存到`TcpConnection`对象中。`onClientMessage()`从TcpConnection对象中取得id，连同数据一起发送给backend，完整实现如下：\n",
    "- [`OnClientMessage()`](./multiplexer/multiplexer_simple.cc#L113)\n",
    "\n",
    "`TcpConnection`的生命期如何管理？由于`client connection`是动态创建并销毁的，其生与灭完全由客户决定，如何保证backend想向它发送数据的时候，这个`TcpConnection`对象还活着？解决思路是用reference counting。当然，不用自己写，用`boost::shared_ptr`即可。`TcpConnection`是muduo中唯一默认采用`shared_ptr`来管理生命期的对象，盖由其动态生命期的本质决定。更多内容请参考第1章。\n",
    "\n",
    "`multiplexer`采用二进制协议，如何测试呢？"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 自动化测试\n",
    "\n",
    "`multiplexer`是muduo网络编程示例中第一个具有non-trivial业务逻辑的网络程序，根据 § 9.7 “分布式程序的自动化回归测试”的思路，我为它编写了测试夹具（test harness）。代码见`examples/multiplexer/harness/`。\n",
    "\n",
    "这个test harness采用Java编写，用的是Netty网络库。这个test harness要同时扮演clients和backend，也就是既要主动发起连接，也要被动接受连接。而且，test harness与`multiplexer`的启动顺序是任意的，如何做到这一点请阅读代码。结构如图7-52所示。\n",
    "\n",
    "<img src=\"./images/7.52.png\" alt=\"7.52\" style=\"zoom:50%;\" />\n",
    "\n",
    "test harness会把各种event汇聚到一个blocking queue里边，方便编写test case。test case则操纵test harness，发起连接、发送数据、检查收到的数据，例如以下是其中一个test case：`testcase/TestOneClientSend.java`。\n",
    "\n",
    "这里的几个test cases都是用Java直接写的，如果有必要，也可以采用Groovy来编写，这样可以在不重启test harness的情况下随时修改、添加test cases。具体做法见笔者的博客[《“过家家”版的移动离线计费系统实现》](https://www.cnblogs.com/Solstice/archive/2011/04/22/2024791.html)。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 将来的改进\n",
    "\n",
    "有了这个自动化的 test harness，我们可以比较方便且安全地修改（甚至重新设计）multiplexer 了。例如：\n",
    "\n",
    "- 增加“backend 发送指令断开 client connection”的功能。有了自动化测试，这个新功能可以被单独测试（开发者测试），而不需要真正的 backend 参与进来。\n",
    "- 将 multiplexer 改用多线程重写。有了自动化回归测试，我们不用担心破坏原有的功能，可以放心大胆地重写。而且由于 test harness 是从外部测试，不是单元测试，重写 multiplexer 的时候不用动 test cases，这样保证了测试的稳定性。另外，这个 test harness 稍加改进还可以进行 stress testing，既可用于验证多线程 multiplexer 的正确性，亦可对比其相对单线程版的效率提升。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 7.13 socks4a代理服务器\n",
    "\n",
    "本节介绍用muduo实现一个简单的socks4a代理服务器([examples/socks4a/](./socks4a/))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## TCP中继器\n",
    "\n",
    "在实现socks4a proxy之前，我们先写一个功能更简单的网络程序—TCP中继器(TCP relay)，或者叫做穷人的tcpdump(poor man’s tcpdump)。一般情况下，客户端程序直接连接服务端，如图7-53所示。\n",
    "\n",
    "```text\n",
    "+---------+                             +---------+\n",
    "| client  |---------------------------->| server  |\n",
    "+---------+                             +---------+\n",
    "```\n",
    "\n",
    "有时候，我们想在client和server之间放一个中继器(relay)，把client与server之间的通信内容记录下来。这时用tcpdump是最方便省事的，但是tcpdump需要root权限，万一拿不到权限呢？穷人有穷人的办法，自己写一个`TcpRelay`，让client连接TcpRelay，再让TcpRelay连接server，如图7-54中的T型结构，TcpRelay扮演了类似proxy的角色。\n",
    "\n",
    "TcpRelay是我们自己写的，可以动动手脚。除了记录通信内容外，还可以制造延时，或者故意翻转1 bit 数据以模拟 router 硬件故障。\n",
    "\n",
    "TcpRelay 的功能（业务逻辑）看上去很简单，无非是把连接 `C` 上收到的数据发给连接 `S`，同时把连接 S 上收到的数据发给连接 `C`。但仔细考虑起来，细节其实不那么简单：\n",
    "1. 建立连接。为了真实模拟 client，TcpRelay 在 accept 连接 `C` 之后才向 server 发起连接 `S`，那么在 `S` 建立起来之前，从 `C` 收到数据怎么办？要不要暂存起来？\n",
    "2. 并发连接的管理。图 7-54 中只画出了一个 client，实际上 TcpRelay 可以服务多个 client，左右两边这些并发连接如何管理，如何防止串话（cross talk）？\n",
    "3. 连接断开。client 和 server 都可能主动断开连接。当 client 主动断开连接 `C` 时，TcpRelay 应该立刻断开 `S`。当 server 主动断开连接 `S` 时，TcpRelay 应立刻断开 C。这样才能比较精确地模拟 client 和 server 的行为。在关闭连接的一刹那，又有新的 client 连接进来，复用了刚刚 close 的 fd 号码，会不会造成串话？万一 client 和 server 几乎同时主动断开连接，TcpRelay 如何应对？\n",
    "4. 速度不匹配。如果连接 `C` 的带宽是 100 kB/s，而连接 `S` 的带宽是 10 MB/s，不巧 server 是个 chargen 服务，会连续发送数据，那么会不会撑爆 TcpRelay 的 buffer？如何限速？特别是在使用 non-blocking IO 和 level-trigger polling 的时候如何限制读取数据的速度？\n",
    "\n",
    "在看 muduo 的实现之前，请读者思考：如果用 Sockets API 来实现 TcpRelay，如何解决以上这些问题。（如果真要实现这么一个功能，可以试试 `splice(2)` 系统调用。）\n",
    "\n",
    "如果用传统多线程阻塞 IO 的方式来实现 TcpRelay 一点也不难，好处是自动解决了速度不匹配的问题，Python 代码如下。这个实现功能上没有问题，但是并发度就高不到哪儿去了。注意以下代码会一个字节一个字节地转发数据，每两个字节之间间隔 1 ms，可用于测试网络程序的消息解码功能（codec）是否完善。\n",
    "\n",
    "<img src=\"./images/7.54.png\" alt=\"7.54\" style=\"zoom:50%;\" />\n",
    "\n",
    "- [tcpreplay.py](../recipes/python/tcprelay.py)\n",
    "\n",
    "TcpRelay的实现很简单，只有几十行代码（[`examples/socks4a/tcprelay.cc`](./socks4a/tcprelay.cc)），主要逻辑都在 `Tunnel class` 里（[`examples/socks4a/tunnel.h`](./socks4a/tunnel.h)）。这个实现很好地解决了前三个问题，第二个问题的解法比较粗暴，用的是 `HighWaterMarkCallback`，如果发送缓冲区堆积的数据大于 10MiB 就断开连接（更好的办法见 § 8.9.3）。TcpRelay 既是服务端，又是客户端，在阅读代码的时候要注意 `onClientMessage()` 处理的是从 server 发来的消息，表示它作为客户端（client）收到的消息，这与前面的 multiplexer 正好相反。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## socks4a代理服务器\n",
    "\n",
    "socks4a的功能与TcpRelay非常相似，也是把连接C上收到的数据发给连接S，同时把连接S上收到的数据发给连接C。它与TcpRelay的区别在于，TcpRelay固定连到某个server地址，而socks4a允许client指定要连哪个server。在accept连接C之后，socks4a server会读几个字节，以了解server的地址，再发起连接S。socks4a的协议非常简单，请参考维基百科。\n",
    "\n",
    "muduo的socks4a代理服务器的实现在[`examples/socks4a/socks4a.cc`](./socks4a/socks4a.cc)，它也使用了`Tunnel class`。与TcpRelay相比，只多了解析server地址这一步骤。目前DNS地址解析这一步用的是阻塞的`gethostbyname()`函数，在真正的系统中，应该换成非阻塞的DNS解析，可参考§ 7.15。muduo的这个socks4a是个标准的网络服务，可以供Web浏览器使用（我正是这么测试它的）。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## N:1 与 1:N 连接转发\n",
    "\n",
    "云风在[《写了一个proxy用途你懂的](https://blog.codingnow.com/2011/05/xtunnel.html)》中写了一个 TCP 隧道 tunnel，程序由三部分组成：N:1 连接转发服务，1:N 连接转发服务，socks 代理服务。\n",
    "\n",
    "我仿照他的思路，用 muduo 实现了这三个程序。不同的是，我没有做数据混淆，所以功能上有所减弱。\n",
    "\n",
    "- N:1 连接转发服务就是 § 7.12 中的 multiplexer（数据选择器）。\n",
    "- 1:N 连接转发服务是云风文中提到的 backend，一个数据分配器（demultiplexer），代码在 [`examples/multiplexer/demux.cc`](./multiplexer/demux.cc)。\n",
    "- socks 代理服务正是 § 7.13.2 实现的 socks4a。\n",
    "\n",
    "有兴趣的读者可以把这三个程序级联起来试一试。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 7.14 短址服务\n",
    "\n",
    "muduo内置了一个简陋的HTTP服务器，可以处理简单的HTTP请求。这个HTTP服务器是面向内网的暴露进程状态的监控端口，不是面向公网的功能完善且健壮的httpd，其接口与J2EE的HttpServlet有几分类似。我们可以拿它来实现一个简单的短URL转发服务，以简要说明其用法。代码位于[examples/shorturl/shorturl.cc](./shorturl/shorturl.cc#L36)。\n",
    "\n",
    "muduo并没有为短连接TCP服务优化，无法发挥多核优势。一种真正高效的优化手段是修改Linux内核，例如Google的[`SO_REUSEPORT`](https://domsch.com/linux/lpc2010/Scaling_techniques_for_servers_with_high_connection%20rates.pdf)内核补丁。\n",
    "\n",
    "读者可以试试建立一个loop转发，例如 “/1” → “/2” → “/3” → “/1”，看看浏览器反应如何。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 7.15 与其他库集成\n",
    "\n",
    "前面介绍的网络应用例子都是直接用 muduo 库收发网络消息，也就是说主要介绍 `TcpConnection`、`TcpServer`、`TcpClient`、`Buffer` 等 class 的使用。本节将稍微深入其内部，介绍 `Channel class` 的用法，通过它可以把其他一些现成的网络库融入 muduo 的 event loop 中。\n",
    "\n",
    "`Channel class` 是 IO 事件回调的分发器（dispatcher），它在 `handleEvent()` 中根据事件的具体类型分别回调 `ReadCallback`、`WriteCallback` 等，代码见 § 8.1.1。每个 `Channel` 对象服务于一个文件描述符，但并不拥有 fd，在析构函数中也不会 `close(fd)`。`Channel` 也使用 muduo 一贯的 `boost::function` 来表示函数回调，[它不是基类](http://www.cppblog.com/Solstice/archive/2012/07/01/181058.aspx)。这样用户代码不必继承 `Channel`，也无需 override 虚函数。\n",
    "\n",
    "- [Channel.h](../muduo/net/Channel.h)\n",
    "\n",
    "Channel与EventLoop的内部交互有两个函数 `EventLoop::updateChannel(Channel*)` 和 `EventLoop::removeChannel(Channel*)`。客户需要在Channel析构前自己调用 `Channel::remove()`。\n",
    "\n",
    "后面我们将通过一些实例来介绍Channel class的使用。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## UDNS\n",
    "\n",
    "[UDNS](https://www.corpit.ru/mjt/udns.html)是一个stub DNS解析器，它能够异步地发起DNS查询，再通过回调函数通知结果。UDNS在设计的时候就考虑到了配合（融入）主程序现有的基于 `select/poll/epoll` 的 event loop 模型，因此它与muduo的配接相对较为容易。由于License限制，本节的代码位于单独的项目中：https://github.com/chenshuo/muduo-udns。\n",
    "\n",
    "> stub的意思是只会查询一个DNS服务器，而不会递归地(recursive)查询多个DNS服务器，因此适合在公司内网使用(http://en.wikipedia.org/wiki/Domain_Name_System#DNS_resolvers)。\n",
    "\n",
    "muduo-udns由三部分组成，一是udns-0.2源码 注；二是UDNS与muduo的配接器(adapter)，即Resolver class，位于Resolver.{h, cc}；三是简单的测试dns.cc，展示Resolver的使用。前两部分构成了muduo-udns程序库。\n",
    "\n",
    "先看`Resolver class`的接口([Resolver.h](../muduo-udns/Resolver.h#L23))：\n",
    "\n",
    "其中第一个构造函数会使用系统默认的DNS服务器地址，第二个构造函数由用户指明DNS服务器的IP地址（见后面的练习1）。用户最关心的是resolve()函数，它会回调用户的`Callback`。\n",
    "\n",
    "在介绍Resolver的实现之前，先来看它的用法([dns.cc](../muduo-udns/dns.cc))，下面这段代码同时解析三个域名，并在`stdout`输出结果。注意回调函数只提供解析后的地址，因此`resolveCallback`需要自己设法记住域名，这里我用的是`boost::bind`。\n",
    "\n",
    "由于是异步解析，因此输出结果的顺序和提交请求的顺序不一定一致，例如：\n",
    "```shell\n",
    "20120822 04:46:39.945033Z 15726 INFO resolved www.google.com -> 74.125.71.104\n",
    "20120822 04:46:41.944464Z 15726 INFO resolved chenshuo.com -> 173.212.209.144\n",
    "20120822 04:46:42.068084Z 15726 INFO resolved www.example.com -> 192.0.43.10\n",
    "```\n",
    "\n",
    "UDNS与muduo Resolver的交互过程如下：\n",
    "1. 初始化 `dns_ctx*` 之后，`Resolver::start()` 调用 `dns_open()` 获得 UDNS 使用的文件描述符，并通过 muduo `Channel` 观察其可读事件。由于 UDNS 始终只用一个 socket fd，只观察一个事件，因此特别容易和现有的 event loop 集成。\n",
    "2. 在解析域名时（`Resolver::resolve()`），调用 `dns_submit_a4()` 发起解析，并通过 `dns_timeouts()` 获得超时的秒数，使用 `EventLoop::runAfter()` 注册单次定时器回调。\n",
    "3. 在 fd 可读时（`Resolver::onRead()`），调用 `dns_ioevent()`。如果 DNS 解析成功，会回调 `Resolver::dns_query_a4()` 通知解析的结果，继而调用 `Resolver::onQueryResult()`，后者会回调用户 Callback。\n",
    "4. 在超时后（`Resolver::onTimer()`），调用 `dns_timeouts()`，必要时继续注册下一次定时器回调。\n",
    "\n",
    "可见UDNS是一个设计良好的库，可与现有的 event loop 很好地结合。UDNS 使用定时器的原因是 UDP 可能丢包，因此程序必须自己处理超时重传。\n",
    "\n",
    "`Resolve class` 不是线程安全的，客户代码只能在 EventLoop 所属的线程调用它的 `Resolver::resolve()` 成员函数，解析结果也是由这个线程回调客户代码。这个函数通过 `loop_->assertInLoopThread();` 来确保不被误用。\n",
    "\n",
    "C++ 程序与 C 语言回调库交互的一个难点在于资源管理，muduo-udns 不得已使用了手工 `new/delete` 的做法，每次解析会在堆上创建 `QueryData` 对象，这样在 UDNS 回调 `Resolver::dns_query_a4()` 时才知道该回调哪个用户 Callback。\n",
    "\n",
    "- 练习 1：补充构造函数 `Resolver(EventLoop* loop, const InetAddress& nameServer)` 的实现。可利用[文档](https://www.corpit.ru/mjt/udns/udns.3.html)介绍的 `dns_add_serv_s()` 函数。\n",
    "- 练习 2：用 muduo-udns 改进 §7.13 的 socks4a 服务器，替换其中阻塞的 `gethostbyname()` 函数调用，实现完全的无阻塞服务。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## c-ares DNS\n",
    "\n",
    "[c-ares DNS](https://c-ares.org/) 注是一款常用的异步DNS解析库，§ 6.2介绍了它的安装方法，本节将简要介绍其与muduo的集成。示例代码位于[`examples/cdns`]，代码结构与 § 7.15.1 的UDNS非常相似。`Resolver.{h, cc}`是 c-ares DNS与muduo的配接器(adapter)，即 `udns::Resolver class；`, [`dns.cc`](./cdns/dns.cc)是简单的测试，展示Resolver的使用。c-ares DNS的选项非常多，本节只是展示其与muduo EventLoop集成的基本做法，`cdns::Resolver`并没有暴露其全部功能。\n",
    "> 功能也比UDNS强大，例如可以读取 `/etc/hosts`。`udns::Resolver` 的构造函数有选项可禁用此功能。\n",
    "\n",
    "`cdns::Resolver`的接口和用法与前面UDNS Resolver相同，只是少了`start()`函数，此处不再重复举例。\n",
    "\n",
    "`cdns::Resolver`的实现与前面UDNS Resolver很相似：\n",
    "1. `Resolver::resolve()`调用`ares_gethostbyname()`发起解析，并通过`ares_timeout()`获得超时的秒数，注册定时器。\n",
    "2. 在fd可读时（`Resolver::onRead()`），调用`ares_process_fd()`。如果DNS解析成功，会回调`Resolver::ares_host_callback()`通知解析的结果，继而调用`Resolver::onQueryResult()`，后者会回调用户`Callback`。\n",
    "> `ares_host_callback()` 相当于前面UDNS的 `dns_query_a4()` 回调。\n",
    "3. 在超时后（`Resolver::onTimer()`），调用`ares_process_fd()`处理这一事件，并再次调用`dns_timeouts()`获得下一次超时的间隔，必要时继续注册下一次定时器回调。\n",
    "\n",
    "与UDNS不同，c-ares DNS会用到不止一个socket文件描述符，而且既会用到fd可读事件，又会用到fd可写事件，因此`cdns::Resolver`的代码比UDNS要复杂一些。`Resolver::ares_sock_create_callback()`是新建socket fd的回调函数，其中会调用`Resolver::onSockCreate()`来创建Channel对象，这正是Resolver没有`start()`成员函数的原因。`Resolver::ares_sock_state_callback()`是变更socket fd状态的回调函数，会通知该观察哪些IO事件（可读and/or可写）。\n",
    "> 因为DNS解析时，如果UDP响应发生消息截断，会改用TCP重发请求\n",
    "\n",
    "练习3：阅读源码并测试c-ares DNS什么时候需要观察“fd可写”事件，然后补充完整`Resolver::onSockStateChange()`。  \n",
    "练习4：修改`Callback`的原型，让Resolver能返回地址列表(`std::vector<InetAddress>`) ，这个练习同样适用于 § 7.15.1的UDNS。  \n",
    "练习5：为[libunbound](https://unbound.docs.nlnetlabs.nl/en/latest/)注编写类似的muduo adapter。注意它似乎没有使用timeout，很奇怪。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## curl\n",
    "\n",
    "libcurl是一个常用的HTTP客户端库，可以方便地下载HTTP和HTTPS数据。libcurl有两套接口，easy和multi，本节介绍的是使用其[multi接口](https://curl.se/libcurl/c/libcurl-multi.html) 注以达到单线程并发访问多个URL的效果。muduo与libcurl搭配的例子见`examples/curl`，其中包含单线程多连接并发下载同一文件的示例，即单线程实现的“多线程下载器”。\n",
    "> 也可以访问FTP服务器\n",
    "\n",
    "libcurl融入muduo EventLoop的复杂度比前面两个DNS库都更高，一方面因为它本身的功能丰富，另一方面也因为它的接口设计更偏重传统阻塞IO（它原本是从curl(1)这个命令行工具剥离出来的），在事件驱动方面的调用、回调、传参都比较烦琐。这里不去详细解释每一个函数的作用，想必读者在读过前两节之后已经对Channel的用法有了基本的了解，对照libcurl文档和muduo代码就能搞明白。\n",
    "\n",
    "- 练习6：修改`curl::Request::DataCallback`的原型，改为以`muduo::net::Buffer*`为参数，方便用户使用。这需要在`curl::Request`中增加`Buffer`成员。\n",
    "\n",
    "第一章我们探讨了多线程程序中的对象生命周期管理技术。在单线程事件驱动的程序中，对象的生命周期管理有时也不简单。比如说图7-55展示的例子，对方断开TCP连接，这个IO事件会触发`Channel::handleEvent()`调用，后者会回调用户提供的`CloseCallback`，而用户代码在`onClose()`中有可能析构`Channel`对象，这就造成了灾难。等于说`Channel::handleEvent()`执行到一半的时候，其所属的`Channel`对象本身被销毁了。这时程序立刻core dump就是最好的结果了。\n",
    "\n",
    "<img src=\"./images/7.55.png\" alt=\"7.55\" style=\"zoom:50%;\" />\n",
    "\n",
    "muduo的解决办法是提供`Channel::tie(const boost::shared_ptr<void>&)`这个函数，用于延长某些对象(可以是Channel对象，也可以是其owner对象)的生命周期，使之长过`Channel::handleEvent()`函数。这也是muduo `TcpConnection`采用`shared_ptr`管理对象生命周期的原因之一；否则的话，`Channel::handleEvent()`有可能引发`TcpConnection`析构，继而把当前Channel对象也析构了，引起程序崩溃。\n",
    "\n",
    "第三方库与muduo集成的另外一个问题是对IO事件变化的理解可能不一致。拿`libcurl`来说，它会在某个文件描述符需要关注的IO事件发生变化的时候通知外围的`event loop`库，比如说原来关注`POLLIN`，现在关注`(POLLIN|POLLOUT)`，muduo在`Curl::socketCallback`回调函数中会相应地调用`Channel::enableWriting()`，能正确处理这种变化。\n",
    "\n",
    "不幸的是，`libcurl`在与`c-ares DNS`配合 注 的时候会出现与muduo不兼容的现象。`libcurl`在访问`URL`的时候要先解析其中的域名，然后再对那个Web服务器发起`TCP`连接。在与`c-ares DNS`搭配时会出现一种情况：`c-ares DNS`解析域名用到的DNS服务器通信的`socket fd1`和`libcurl`对Web服务器发起`TCP`连接的fd恰好相等，即`fd1==fd2`。原因是`POSIX`操作系统总是选择当前最小可用的文件描述符，当DNS解析完成后，`libcurl`内部使用的`c-ares DNS`会关闭`fd1`，`libcurl`随后马上新建一个`TCP socket` `fd2`，它有可能恰好复用了`fd1`的值。\n",
    "> 这两个库是同一个作者，libcurl默认会用`gethostbyname()`执行同步DNS解析，在有c-ares DNS的时候它执行异步DNS解析。\n",
    "\n",
    "但这时`libcurl`不会认为文件描述符或其关注的IO事件发生了变化，也就不会通知muduo去销毁并新建Channel对象。这种做法与传统的基于`select(2)`和`poll(2)`的`event loop`配合不会有问题，因为`select(2)`和`poll(2)`是上下文无关的，每次都从输入重建要关注的文件描述符列表。但是在与`epoll(4)`配合的时候就有问题了，关闭`fd1`会使得epoll从关注列表(`watch list`)中移除`fd1`的条目，新建的同名`fd2`却没有机会加入IO事件watch list，也就不会收到任何IO事件通知。这个问题无法在muduo内部修复，只能修改上游的程序库。\n",
    "\n",
    "另外一个问题是`libcurl`在通知muduo取消关注某个`fd`的时候已经先关闭了它，这将造成muduo调用：`epoll_ctl(epollfd_, EPOLL_CTL_DEL, fd, NULL)`时会返回错误，因为关闭文件描述符已经就地把epoll watch list中除掉了。为了应对这种情况，我不得不更改了`EPollPoller::update()`的错误处理，放宽检查。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 更多\n",
    "\n",
    "除了前面举的几个例子，muduo当然还可以将其他涉及网络IO的库融入其EventLoop/Channel框架，我能想到的有：\n",
    "\n",
    "- libmicrohttpd— 可嵌入的HTTP服务器。\n",
    "- libpq— PostgreSQL的官方客户端库。\n",
    "- libdrizzle— MySQL的非官方客户端库。\n",
    "- QuickFIX— 常用的FIX消息库。\n",
    "\n",
    "在有具体应用场景的时候，我多半会为之提供muduo adapter，也欢迎用户贡献有关补丁。\n",
    "\n",
    "另外一个扩展思路是，对每个TCP连接创建一个lua state，用muduo为lua提供通信机制。然后用lua来编写业务逻辑，这也可以做到在线更改逻辑而不重启进程。就像OpenResty 注 和云风的skynet 注 那样。这种做法还可以利用coroutine来简化业务逻辑的实现。"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "C++17",
   "language": "C++17",
   "name": "xcpp17"
  },
  "language_info": {
   "codemirror_mode": "text/x-c++src",
   "file_extension": ".cpp",
   "mimetype": "text/x-c++src",
   "name": "C++17",
   "version": "17"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
